// SPDX-License-Identifier: GPL-2.0-only/* drbd_worker.c This file is part of DRBD by Philipp Reisner and Lars Ellenberg. Copyright (C) 2001-2008, LINBIT Information Technologies GmbH. Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>. Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>. */ #include <linux/module.h>#include <linux/drbd.h>#include <linux/sched/signal.h>#include <linux/wait.h>#include <linux/mm.h>#include <linux/memcontrol.h>#include <linux/mm_inline.h>#include <linux/slab.h>#include <linux/random.h>#include <linux/string.h>#include <linux/scatterlist.h>#include <linux/part_stat.h> #include "drbd_int.h"#include "drbd_protocol.h"#include "drbd_req.h" static int make_ov_request(struct drbd_peer_device *, int);static int make_resync_request(struct drbd_peer_device *, int); /* endio handlers: * drbd_md_endio (defined here) * drbd_request_endio (defined here) * drbd_peer_request_endio (defined here) * drbd_bm_endio (defined in drbd_bitmap.c) * * For all these callbacks, note the following: * The callbacks will be called in irq context by the IDE drivers, * and in Softirqs/Tasklets/BH context by the SCSI drivers. * Try to get the locking right :) * */ /* used for synchronous meta data and bitmap IO * submitted by drbd_md_sync_page_io() */void drbd_md_endio(struct bio *bio){ struct drbd_device *device; device = bio->bi_private; device->md_io.error = blk_status_to_errno(bio->bi_status); /* special case: drbd_md_read() during drbd_adm_attach() */ if (device->ldev) put_ldev(device); bio_put(bio); /* We grabbed an extra reference in _drbd_md_sync_page_io() to be able * to timeout on the lower level device, and eventually detach from it. * If this io completion runs after that timeout expired, this * drbd_md_put_buffer() may allow us to finally try and re-attach. * During normal operation, this only puts that extra reference * down to 1 again. * Make sure we first drop the reference, and only then signal * completion, or we may (in drbd_al_read_log()) cycle so fast into the * next drbd_md_sync_page_io(), that we trigger the * ASSERT(atomic_read(&device->md_io_in_use) == 1) there. */ drbd_md_put_buffer(device); device->md_io.done = 1; wake_up(&device->misc_wait);} /* reads on behalf of the partner, * "submitted" by the receiver */static void drbd_endio_read_sec_final(struct drbd_peer_request *peer_req) __releases(local){ unsigned long flags = 0; struct drbd_peer_device *peer_device = peer_req->peer_device; struct drbd_device *device = peer_device->device; spin_lock_irqsave(&device->resource->req_lock, flags); device->read_cnt += peer_req->i.size >> 9; list_del(&peer_req->w.list); if (list_empty(&device->read_ee)) wake_up(&device->ee_wait); if (test_bit(__EE_WAS_ERROR, &peer_req->flags)) __drbd_chk_io_error(device, DRBD_READ_ERROR); spin_unlock_irqrestore(&device->resource->req_lock, flags); drbd_queue_work(&peer_device->connection->sender_work, &peer_req->w); put_ldev(device);} /* writes on behalf of the partner, or resync writes, * "submitted" by the receiver, final stage. */void drbd_endio_write_sec_final(struct drbd_peer_request *peer_req) __releases(local){ unsigned long flags = 0; struct drbd_peer_device *peer_device = peer_req->peer_device; struct drbd_device *device = peer_device->device; struct drbd_connection *connection = peer_device->connection; struct drbd_interval i; int do_wake; u64 block_id; int do_al_complete_io; /* after we moved peer_req to done_ee, * we may no longer access it, * it may be freed/reused already! * (as soon as we release the req_lock) */ i = peer_req->i; do_al_complete_io = peer_req->flags & EE_CALL_AL_COMPLETE_IO; block_id = peer_req->block_id; peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO; if (peer_req->flags & EE_WAS_ERROR) { /* In protocol != C, we usually do not send write acks. * In case of a write error, send the neg ack anyways. */ if (!__test_and_set_bit(__EE_SEND_WRITE_ACK, &peer_req->flags)) inc_unacked(device); drbd_set_out_of_sync(peer_device, peer_req->i.sector, peer_req->i.size); } spin_lock_irqsave(&device->resource->req_lock, flags); device->writ_cnt += peer_req->i.size >> 9; list_move_tail(&peer_req->w.list, &device->done_ee); /* * Do not remove from the write_requests tree here: we did not send the * Ack yet and did not wake possibly waiting conflicting requests. * Removed from the tree from "drbd_process_done_ee" within the * appropriate dw.cb (e_end_block/e_end_resync_block) or from * _drbd_clear_done_ee. */ do_wake = list_empty(block_id == ID_SYNCER ? &device->sync_ee : &device->active_ee); /* FIXME do we want to detach for failed REQ_OP_DISCARD? * ((peer_req->flags & (EE_WAS_ERROR|EE_TRIM)) == EE_WAS_ERROR) */ if (peer_req->flags & EE_WAS_ERROR) __drbd_chk_io_error(device, DRBD_WRITE_ERROR); if (connection->cstate >= C_WF_REPORT_PARAMS) { kref_get(&device->kref); /* put is in drbd_send_acks_wf() */ if (!queue_work(connection->ack_sender, &peer_device->send_acks_work)) kref_put(&device->kref, drbd_destroy_device); } spin_unlock_irqrestore(&device->resource->req_lock, flags); if (block_id == ID_SYNCER) drbd_rs_complete_io(device, i.sector); if (do_wake) wake_up(&device->ee_wait); if (do_al_complete_io) drbd_al_complete_io(device, &i); put_ldev(device);} /* writes on behalf of the partner, or resync writes, * "submitted" by the receiver. */void drbd_peer_request_endio(struct bio *bio){ struct drbd_peer_request *peer_req = bio->bi_private; struct drbd_device *device = peer_req->peer_device->device; bool is_write = bio_data_dir(bio) == WRITE; bool is_discard = bio_op(bio) == REQ_OP_WRITE_ZEROES || bio_op(bio) == REQ_OP_DISCARD; if (bio->bi_status && drbd_ratelimit()) drbd_warn(device, "%s: error=%d s=%llus\n", is_write ? (is_discard ? "discard" : "write") : "read", bio->bi_status, (unsigned long long)peer_req->i.sector); if (bio->bi_status) set_bit(__EE_WAS_ERROR, &peer_req->flags); bio_put(bio); /* no need for the bio anymore */ if (atomic_dec_and_test(&peer_req->pending_bios)) { if (is_write) drbd_endio_write_sec_final(peer_req); else drbd_endio_read_sec_final(peer_req); }} static voiddrbd_panic_after_delayed_completion_of_aborted_request(struct drbd_device *device){ panic("drbd%u %s/%u potential random memory corruption caused by delayed completion of aborted local request\n", device->minor, device->resource->name, device->vnr);} /* read, readA or write requests on R_PRIMARY coming from drbd_make_request */void drbd_request_endio(struct bio *bio){ unsigned long flags; struct drbd_request *req = bio->bi_private; struct drbd_device *device = req->device; struct bio_and_error m; enum drbd_req_event what; /* If this request was aborted locally before, * but now was completed "successfully", * chances are that this caused arbitrary data corruption. * * "aborting" requests, or force-detaching the disk, is intended for * completely blocked/hung local backing devices which do no longer * complete requests at all, not even do error completions. In this * situation, usually a hard-reset and failover is the only way out. * * By "aborting", basically faking a local error-completion, * we allow for a more graceful swichover by cleanly migrating services. * Still the affected node has to be rebooted "soon". * * By completing these requests, we allow the upper layers to re-use * the associated data pages. * * If later the local backing device "recovers", and now DMAs some data * from disk into the original request pages, in the best case it will * just put random data into unused pages; but typically it will corrupt * meanwhile completely unrelated data, causing all sorts of damage. * * Which means delayed successful completion, * especially for READ requests, * is a reason to panic(). * * We assume that a delayed *error* completion is OK, * though we still will complain noisily about it. */ if (unlikely(req->rq_state & RQ_LOCAL_ABORTED)) { if (drbd_ratelimit()) drbd_emerg(device, "delayed completion of aborted local request; disk-timeout may be too aggressive\n"); if (!bio->bi_status) drbd_panic_after_delayed_completion_of_aborted_request(device); } /* to avoid recursion in __req_mod */ if (unlikely(bio->bi_status)) { switch (bio_op(bio)) { case REQ_OP_WRITE_ZEROES: case REQ_OP_DISCARD: if (bio->bi_status == BLK_STS_NOTSUPP) what = DISCARD_COMPLETED_NOTSUPP; else what = DISCARD_COMPLETED_WITH_ERROR; break; case REQ_OP_READ: if (bio->bi_opf & REQ_RAHEAD) what = READ_AHEAD_COMPLETED_WITH_ERROR; else what = READ_COMPLETED_WITH_ERROR; break; default: what = WRITE_COMPLETED_WITH_ERROR; break; } } else { what = COMPLETED_OK; } req->private_bio = ERR_PTR(blk_status_to_errno(bio->bi_status)); bio_put(bio); /* not req_mod(), we need irqsave here! */ spin_lock_irqsave(&device->resource->req_lock, flags); __req_mod(req, what, NULL, &m); spin_unlock_irqrestore(&device->resource->req_lock, flags); put_ldev(device); if (m.bio) complete_master_bio(device, &m);} void drbd_csum_ee(struct crypto_shash *tfm, struct drbd_peer_request *peer_req, void *digest){ SHASH_DESC_ON_STACK(desc, tfm); struct page *page = peer_req->pages; struct page *tmp; unsigned len; void *src; desc->tfm = tfm; crypto_shash_init(desc); src = kmap_atomic(page); while ((tmp = page_chain_next(page))) { /* all but the last page will be fully used */ crypto_shash_update(desc, src, PAGE_SIZE); kunmap_atomic(src); page = tmp; src = kmap_atomic(page); } /* and now the last, possibly only partially used page */ len = peer_req->i.size & (PAGE_SIZE - 1); crypto_shash_update(desc, src, len ?: PAGE_SIZE); kunmap_atomic(src); crypto_shash_final(desc, digest); shash_desc_zero(desc);} void drbd_csum_bio(struct crypto_shash *tfm, struct bio *bio, void *digest){ SHASH_DESC_ON_STACK(desc, tfm); struct bio_vec bvec; struct bvec_iter iter; desc->tfm = tfm; crypto_shash_init(desc); bio_for_each_segment(bvec, bio, iter) { u8 *src; src = bvec_kmap_local(&bvec); crypto_shash_update(desc, src, bvec.bv_len); kunmap_local(src); } crypto_shash_final(desc, digest); shash_desc_zero(desc);} /* MAYBE merge common code with w_e_end_ov_req */static int w_e_send_csum(struct drbd_work *w, int cancel){ struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w); struct drbd_peer_device *peer_device = peer_req->peer_device; struct drbd_device *device = peer_device->device; int digest_size; void *digest; int err = 0; if (unlikely(cancel)) goto out; if (unlikely((peer_req->flags & EE_WAS_ERROR) != 0)) goto out; digest_size = crypto_shash_digestsize(peer_device->connection->csums_tfm); digest = kmalloc(digest_size, GFP_NOIO); if (digest) { sector_t sector = peer_req->i.sector; unsigned int size = peer_req->i.size; drbd_csum_ee(peer_device->connection->csums_tfm, peer_req, digest); /* Free peer_req and pages before send. * In case we block on congestion, we could otherwise run into * some distributed deadlock, if the other side blocks on * congestion as well, because our receiver blocks in * drbd_alloc_pages due to pp_in_use > max_buffers. */ drbd_free_peer_req(device, peer_req); peer_req = NULL; inc_rs_pending(peer_device); err = drbd_send_drequest_csum(peer_device, sector, size, digest, digest_size, P_CSUM_RS_REQUEST); kfree(digest); } else { drbd_err(device, "kmalloc() of digest failed.\n"); err = -ENOMEM; } out: if (peer_req) drbd_free_peer_req(device, peer_req); if (unlikely(err)) drbd_err(device, "drbd_send_drequest(..., csum) failed\n"); return err;} #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN) static int read_for_csum(struct drbd_peer_device *peer_device, sector_t sector, int size){ struct drbd_device *device = peer_device->device; struct drbd_peer_request *peer_req; if (!get_ldev(device)) return -EIO; /* GFP_TRY, because if there is no memory available right now, this may * be rescheduled for later. It is "only" background resync, after all. */ peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER /* unused */, sector, size, size, GFP_TRY); if (!peer_req) goto defer; peer_req->w.cb = w_e_send_csum; peer_req->opf = REQ_OP_READ; spin_lock_irq(&device->resource->req_lock); list_add_tail(&peer_req->w.list, &device->read_ee); spin_unlock_irq(&device->resource->req_lock); atomic_add(size >> 9, &device->rs_sect_ev); if (drbd_submit_peer_request(peer_req) == 0) return 0; /* If it failed because of ENOMEM, retry should help. If it failed * because bio_add_page failed (probably broken lower level driver), * retry may or may not help. * If it does not, you may need to force disconnect. */ spin_lock_irq(&device->resource->req_lock); list_del(&peer_req->w.list); spin_unlock_irq(&device->resource->req_lock); drbd_free_peer_req(device, peer_req);defer: put_ldev(device); return -EAGAIN;} int w_resync_timer(struct drbd_work *w, int cancel){ struct drbd_device *device = container_of(w, struct drbd_device, resync_work); switch (device->state.conn) { case C_VERIFY_S: make_ov_request(first_peer_device(device), cancel); break; case C_SYNC_TARGET: make_resync_request(first_peer_device(device), cancel); break; } return 0;} void resync_timer_fn(struct timer_list *t){ struct drbd_device *device = timer_container_of(device, t, resync_timer); drbd_queue_work_if_unqueued( &first_peer_device(device)->connection->sender_work, &device->resync_work);} static void fifo_set(struct fifo_buffer *fb, int value){ int i; for (i = 0; i < fb->size; i++) fb->values[i] = value;} static int fifo_push(struct fifo_buffer *fb, int value){ int ov; ov = fb->values[fb->head_index]; fb->values[fb->head_index++] = value; if (fb->head_index >= fb->size) fb->head_index = 0; return ov;} static void fifo_add_val(struct fifo_buffer *fb, int value){ int i; for (i = 0; i < fb->size; i++) fb->values[i] += value;} struct fifo_buffer *fifo_alloc(unsigned int fifo_size){ struct fifo_buffer *fb; fb = kzalloc_flex(*fb, values, fifo_size, GFP_NOIO); if (!fb) return NULL; fb->head_index = 0; fb->size = fifo_size; fb->total = 0; return fb;} static int drbd_rs_controller(struct drbd_peer_device *peer_device, unsigned int sect_in){ struct drbd_device *device = peer_device->device; struct disk_conf *dc;Lines 1–500 of 2223. Load more
// SPDX-License-Identifier: GPL-2.0-only /* drbd_worker.c This file is part of DRBD by Philipp Reisner and Lars Ellenberg. Copyright (C) 2001-2008, LINBIT Information Technologies GmbH. Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>. Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>. */ #include <linux/module.h> #include <linux/drbd.h> #include <linux/sched/signal.h> #include <linux/wait.h> #include <linux/mm.h> #include <linux/memcontrol.h> #include <linux/mm_inline.h> #include <linux/slab.h> #include <linux/random.h> #include <linux/string.h> #include <linux/scatterlist.h> #include <linux/part_stat.h> #include "drbd_int.h" #include "drbd_protocol.h" #include "drbd_req.h" static int make_ov_request(struct drbd_peer_device *, int); static int make_resync_request(struct drbd_peer_device *, int); /* endio handlers: * drbd_md_endio (defined here) * drbd_request_endio (defined here) * drbd_peer_request_endio (defined here) * drbd_bm_endio (defined in drbd_bitmap.c) * * For all these callbacks, note the following: * The callbacks will be called in irq context by the IDE drivers, * and in Softirqs/Tasklets/BH context by the SCSI drivers. * Try to get the locking right :) * */ /* used for synchronous meta data and bitmap IO * submitted by drbd_md_sync_page_io() */ void drbd_md_endio(struct bio *bio) { struct drbd_device *device; device = bio->bi_private; device->md_io.error = blk_status_to_errno(bio->bi_status); /* special case: drbd_md_read() during drbd_adm_attach() */ if (device->ldev) put_ldev(device); bio_put(bio); /* We grabbed an extra reference in _drbd_md_sync_page_io() to be able * to timeout on the lower level device, and eventually detach from it. * If this io completion runs after that timeout expired, this * drbd_md_put_buffer() may allow us to finally try and re-attach. * During normal operation, this only puts that extra reference * down to 1 again. * Make sure we first drop the reference, and only then signal * completion, or we may (in drbd_al_read_log()) cycle so fast into the * next drbd_md_sync_page_io(), that we trigger the * ASSERT(atomic_read(&device->md_io_in_use) == 1) there. */ drbd_md_put_buffer(device); device->md_io.done = 1; wake_up(&device->misc_wait); } /* reads on behalf of the partner, * "submitted" by the receiver */ static void drbd_endio_read_sec_final(struct drbd_peer_request *peer_req) __releases(local) { unsigned long flags = 0; struct drbd_peer_device *peer_device = peer_req->peer_device; struct drbd_device *device = peer_device->device; spin_lock_irqsave(&device->resource->req_lock, flags); device->read_cnt += peer_req->i.size >> 9; list_del(&peer_req->w.list); if (list_empty(&device->read_ee)) wake_up(&device->ee_wait); if (test_bit(__EE_WAS_ERROR, &peer_req->flags)) __drbd_chk_io_error(device, DRBD_READ_ERROR); spin_unlock_irqrestore(&device->resource->req_lock, flags); drbd_queue_work(&peer_device->connection->sender_work, &peer_req->w); put_ldev(device); } /* writes on behalf of the partner, or resync writes, * "submitted" by the receiver, final stage. */ void drbd_endio_write_sec_final(struct drbd_peer_request *peer_req) __releases(local) { unsigned long flags = 0; struct drbd_peer_device *peer_device = peer_req->peer_device; struct drbd_device *device = peer_device->device; struct drbd_connection *connection = peer_device->connection; struct drbd_interval i; int do_wake; u64 block_id; int do_al_complete_io; /* after we moved peer_req to done_ee, * we may no longer access it, * it may be freed/reused already! * (as soon as we release the req_lock) */ i = peer_req->i; do_al_complete_io = peer_req->flags & EE_CALL_AL_COMPLETE_IO; block_id = peer_req->block_id; peer_req->flags &= ~EE_CALL_AL_COMPLETE_IO; if (peer_req->flags & EE_WAS_ERROR) { /* In protocol != C, we usually do not send write acks. * In case of a write error, send the neg ack anyways. */ if (!__test_and_set_bit(__EE_SEND_WRITE_ACK, &peer_req->flags)) inc_unacked(device); drbd_set_out_of_sync(peer_device, peer_req->i.sector, peer_req->i.size); } spin_lock_irqsave(&device->resource->req_lock, flags); device->writ_cnt += peer_req->i.size >> 9; list_move_tail(&peer_req->w.list, &device->done_ee); /* * Do not remove from the write_requests tree here: we did not send the * Ack yet and did not wake possibly waiting conflicting requests. * Removed from the tree from "drbd_process_done_ee" within the * appropriate dw.cb (e_end_block/e_end_resync_block) or from * _drbd_clear_done_ee. */ do_wake = list_empty(block_id == ID_SYNCER ? &device->sync_ee : &device->active_ee); /* FIXME do we want to detach for failed REQ_OP_DISCARD? * ((peer_req->flags & (EE_WAS_ERROR|EE_TRIM)) == EE_WAS_ERROR) */ if (peer_req->flags & EE_WAS_ERROR) __drbd_chk_io_error(device, DRBD_WRITE_ERROR); if (connection->cstate >= C_WF_REPORT_PARAMS) { kref_get(&device->kref); /* put is in drbd_send_acks_wf() */ if (!queue_work(connection->ack_sender, &peer_device->send_acks_work)) kref_put(&device->kref, drbd_destroy_device); } spin_unlock_irqrestore(&device->resource->req_lock, flags); if (block_id == ID_SYNCER) drbd_rs_complete_io(device, i.sector); if (do_wake) wake_up(&device->ee_wait); if (do_al_complete_io) drbd_al_complete_io(device, &i); put_ldev(device); } /* writes on behalf of the partner, or resync writes, * "submitted" by the receiver. */ void drbd_peer_request_endio(struct bio *bio) { struct drbd_peer_request *peer_req = bio->bi_private; struct drbd_device *device = peer_req->peer_device->device; bool is_write = bio_data_dir(bio) == WRITE; bool is_discard = bio_op(bio) == REQ_OP_WRITE_ZEROES || bio_op(bio) == REQ_OP_DISCARD; if (bio->bi_status && drbd_ratelimit()) drbd_warn(device, "%s: error=%d s=%llus\n", is_write ? (is_discard ? "discard" : "write") : "read", bio->bi_status, (unsigned long long)peer_req->i.sector); if (bio->bi_status) set_bit(__EE_WAS_ERROR, &peer_req->flags); bio_put(bio); /* no need for the bio anymore */ if (atomic_dec_and_test(&peer_req->pending_bios)) { if (is_write) drbd_endio_write_sec_final(peer_req); else drbd_endio_read_sec_final(peer_req); } } static void drbd_panic_after_delayed_completion_of_aborted_request(struct drbd_device *device) { panic("drbd%u %s/%u potential random memory corruption caused by delayed completion of aborted local request\n", device->minor, device->resource->name, device->vnr); } /* read, readA or write requests on R_PRIMARY coming from drbd_make_request */ void drbd_request_endio(struct bio *bio) { unsigned long flags; struct drbd_request *req = bio->bi_private; struct drbd_device *device = req->device; struct bio_and_error m; enum drbd_req_event what; /* If this request was aborted locally before, * but now was completed "successfully", * chances are that this caused arbitrary data corruption. * * "aborting" requests, or force-detaching the disk, is intended for * completely blocked/hung local backing devices which do no longer * complete requests at all, not even do error completions. In this * situation, usually a hard-reset and failover is the only way out. * * By "aborting", basically faking a local error-completion, * we allow for a more graceful swichover by cleanly migrating services. * Still the affected node has to be rebooted "soon". * * By completing these requests, we allow the upper layers to re-use * the associated data pages. * * If later the local backing device "recovers", and now DMAs some data * from disk into the original request pages, in the best case it will * just put random data into unused pages; but typically it will corrupt * meanwhile completely unrelated data, causing all sorts of damage. * * Which means delayed successful completion, * especially for READ requests, * is a reason to panic(). * * We assume that a delayed *error* completion is OK, * though we still will complain noisily about it. */ if (unlikely(req->rq_state & RQ_LOCAL_ABORTED)) { if (drbd_ratelimit()) drbd_emerg(device, "delayed completion of aborted local request; disk-timeout may be too aggressive\n"); if (!bio->bi_status) drbd_panic_after_delayed_completion_of_aborted_request(device); } /* to avoid recursion in __req_mod */ if (unlikely(bio->bi_status)) { switch (bio_op(bio)) { case REQ_OP_WRITE_ZEROES: case REQ_OP_DISCARD: if (bio->bi_status == BLK_STS_NOTSUPP) what = DISCARD_COMPLETED_NOTSUPP; else what = DISCARD_COMPLETED_WITH_ERROR; break; case REQ_OP_READ: if (bio->bi_opf & REQ_RAHEAD) what = READ_AHEAD_COMPLETED_WITH_ERROR; else what = READ_COMPLETED_WITH_ERROR; break; default: what = WRITE_COMPLETED_WITH_ERROR; break; } } else { what = COMPLETED_OK; } req->private_bio = ERR_PTR(blk_status_to_errno(bio->bi_status)); bio_put(bio); /* not req_mod(), we need irqsave here! */ spin_lock_irqsave(&device->resource->req_lock, flags); __req_mod(req, what, NULL, &m); spin_unlock_irqrestore(&device->resource->req_lock, flags); put_ldev(device); if (m.bio) complete_master_bio(device, &m); } void drbd_csum_ee(struct crypto_shash *tfm, struct drbd_peer_request *peer_req, void *digest) { SHASH_DESC_ON_STACK(desc, tfm); struct page *page = peer_req->pages; struct page *tmp; unsigned len; void *src; desc->tfm = tfm; crypto_shash_init(desc); src = kmap_atomic(page); while ((tmp = page_chain_next(page))) { /* all but the last page will be fully used */ crypto_shash_update(desc, src, PAGE_SIZE); kunmap_atomic(src); page = tmp; src = kmap_atomic(page); } /* and now the last, possibly only partially used page */ len = peer_req->i.size & (PAGE_SIZE - 1); crypto_shash_update(desc, src, len ?: PAGE_SIZE); kunmap_atomic(src); crypto_shash_final(desc, digest); shash_desc_zero(desc); } void drbd_csum_bio(struct crypto_shash *tfm, struct bio *bio, void *digest) { SHASH_DESC_ON_STACK(desc, tfm); struct bio_vec bvec; struct bvec_iter iter; desc->tfm = tfm; crypto_shash_init(desc); bio_for_each_segment(bvec, bio, iter) { u8 *src; src = bvec_kmap_local(&bvec); crypto_shash_update(desc, src, bvec.bv_len); kunmap_local(src); } crypto_shash_final(desc, digest); shash_desc_zero(desc); } /* MAYBE merge common code with w_e_end_ov_req */ static int w_e_send_csum(struct drbd_work *w, int cancel) { struct drbd_peer_request *peer_req = container_of(w, struct drbd_peer_request, w); struct drbd_peer_device *peer_device = peer_req->peer_device; struct drbd_device *device = peer_device->device; int digest_size; void *digest; int err = 0; if (unlikely(cancel)) goto out; if (unlikely((peer_req->flags & EE_WAS_ERROR) != 0)) goto out; digest_size = crypto_shash_digestsize(peer_device->connection->csums_tfm); digest = kmalloc(digest_size, GFP_NOIO); if (digest) { sector_t sector = peer_req->i.sector; unsigned int size = peer_req->i.size; drbd_csum_ee(peer_device->connection->csums_tfm, peer_req, digest); /* Free peer_req and pages before send. * In case we block on congestion, we could otherwise run into * some distributed deadlock, if the other side blocks on * congestion as well, because our receiver blocks in * drbd_alloc_pages due to pp_in_use > max_buffers. */ drbd_free_peer_req(device, peer_req); peer_req = NULL; inc_rs_pending(peer_device); err = drbd_send_drequest_csum(peer_device, sector, size, digest, digest_size, P_CSUM_RS_REQUEST); kfree(digest); } else { drbd_err(device, "kmalloc() of digest failed.\n"); err = -ENOMEM; } out: if (peer_req) drbd_free_peer_req(device, peer_req); if (unlikely(err)) drbd_err(device, "drbd_send_drequest(..., csum) failed\n"); return err; } #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN) static int read_for_csum(struct drbd_peer_device *peer_device, sector_t sector, int size) { struct drbd_device *device = peer_device->device; struct drbd_peer_request *peer_req; if (!get_ldev(device)) return -EIO; /* GFP_TRY, because if there is no memory available right now, this may * be rescheduled for later. It is "only" background resync, after all. */ peer_req = drbd_alloc_peer_req(peer_device, ID_SYNCER /* unused */, sector, size, size, GFP_TRY); if (!peer_req) goto defer; peer_req->w.cb = w_e_send_csum; peer_req->opf = REQ_OP_READ; spin_lock_irq(&device->resource->req_lock); list_add_tail(&peer_req->w.list, &device->read_ee); spin_unlock_irq(&device->resource->req_lock); atomic_add(size >> 9, &device->rs_sect_ev); if (drbd_submit_peer_request(peer_req) == 0) return 0; /* If it failed because of ENOMEM, retry should help. If it failed * because bio_add_page failed (probably broken lower level driver), * retry may or may not help. * If it does not, you may need to force disconnect. */ spin_lock_irq(&device->resource->req_lock); list_del(&peer_req->w.list); spin_unlock_irq(&device->resource->req_lock); drbd_free_peer_req(device, peer_req); defer: put_ldev(device); return -EAGAIN; } int w_resync_timer(struct drbd_work *w, int cancel) { struct drbd_device *device = container_of(w, struct drbd_device, resync_work); switch (device->state.conn) { case C_VERIFY_S: make_ov_request(first_peer_device(device), cancel); break; case C_SYNC_TARGET: make_resync_request(first_peer_device(device), cancel); break; } return 0; } void resync_timer_fn(struct timer_list *t) { struct drbd_device *device = timer_container_of(device, t, resync_timer); drbd_queue_work_if_unqueued( &first_peer_device(device)->connection->sender_work, &device->resync_work); } static void fifo_set(struct fifo_buffer *fb, int value) { int i; for (i = 0; i < fb->size; i++) fb->values[i] = value; } static int fifo_push(struct fifo_buffer *fb, int value) { int ov; ov = fb->values[fb->head_index]; fb->values[fb->head_index++] = value; if (fb->head_index >= fb->size) fb->head_index = 0; return ov; } static void fifo_add_val(struct fifo_buffer *fb, int value) { int i; for (i = 0; i < fb->size; i++) fb->values[i] += value; } struct fifo_buffer *fifo_alloc(unsigned int fifo_size) { struct fifo_buffer *fb; fb = kzalloc_flex(*fb, values, fifo_size, GFP_NOIO); if (!fb) return NULL; fb->head_index = 0; fb->size = fifo_size; fb->total = 0; return fb; } static int drbd_rs_controller(struct drbd_peer_device *peer_device, unsigned int sect_in) { struct drbd_device *device = peer_device->device; struct disk_conf *dc;