// SPDX-License-Identifier: GPL-2.0-only/* drbd_req.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/slab.h>#include <linux/drbd.h>#include "drbd_int.h"#include "drbd_req.h" static bool drbd_may_do_local_read(struct drbd_device *device, sector_t sector, int size); static struct drbd_request *drbd_req_new(struct drbd_device *device, struct bio *bio_src){ struct drbd_request *req; req = mempool_alloc(&drbd_request_mempool, GFP_NOIO); if (!req) return NULL; memset(req, 0, sizeof(*req)); req->rq_state = (bio_data_dir(bio_src) == WRITE ? RQ_WRITE : 0) | (bio_op(bio_src) == REQ_OP_WRITE_ZEROES ? RQ_ZEROES : 0) | (bio_op(bio_src) == REQ_OP_DISCARD ? RQ_UNMAP : 0); req->device = device; req->master_bio = bio_src; req->epoch = 0; drbd_clear_interval(&req->i); req->i.sector = bio_src->bi_iter.bi_sector; req->i.size = bio_src->bi_iter.bi_size; req->i.local = true; req->i.waiting = false; INIT_LIST_HEAD(&req->tl_requests); INIT_LIST_HEAD(&req->w.list); INIT_LIST_HEAD(&req->req_pending_master_completion); INIT_LIST_HEAD(&req->req_pending_local); /* one reference to be put by __drbd_make_request */ atomic_set(&req->completion_ref, 1); /* one kref as long as completion_ref > 0 */ kref_init(&req->kref); return req;} static void drbd_remove_request_interval(struct rb_root *root, struct drbd_request *req){ struct drbd_device *device = req->device; struct drbd_interval *i = &req->i; drbd_remove_interval(root, i); /* Wake up any processes waiting for this request to complete. */ if (i->waiting) wake_up(&device->misc_wait);} void drbd_req_destroy(struct kref *kref){ struct drbd_request *req = container_of(kref, struct drbd_request, kref); struct drbd_device *device = req->device; const unsigned s = req->rq_state; if ((req->master_bio && !(s & RQ_POSTPONED)) || atomic_read(&req->completion_ref) || (s & RQ_LOCAL_PENDING) || ((s & RQ_NET_MASK) && !(s & RQ_NET_DONE))) { drbd_err(device, "drbd_req_destroy: Logic BUG rq_state = 0x%x, completion_ref = %d\n", s, atomic_read(&req->completion_ref)); return; } /* If called from mod_rq_state (expected normal case) or * drbd_send_and_submit (the less likely normal path), this holds the * req_lock, and req->tl_requests will typicaly be on ->transfer_log, * though it may be still empty (never added to the transfer log). * * If called from do_retry(), we do NOT hold the req_lock, but we are * still allowed to unconditionally list_del(&req->tl_requests), * because it will be on a local on-stack list only. */ list_del_init(&req->tl_requests); /* finally remove the request from the conflict detection * respective block_id verification interval tree. */ if (!drbd_interval_empty(&req->i)) { struct rb_root *root; if (s & RQ_WRITE) root = &device->write_requests; else root = &device->read_requests; drbd_remove_request_interval(root, req); } else if (s & (RQ_NET_MASK & ~RQ_NET_DONE) && req->i.size != 0) drbd_err(device, "drbd_req_destroy: Logic BUG: interval empty, but: rq_state=0x%x, sect=%llu, size=%u\n", s, (unsigned long long)req->i.sector, req->i.size); /* if it was a write, we may have to set the corresponding * bit(s) out-of-sync first. If it had a local part, we need to * release the reference to the activity log. */ if (s & RQ_WRITE) { /* Set out-of-sync unless both OK flags are set * (local only or remote failed). * Other places where we set out-of-sync: * READ with local io-error */ /* There is a special case: * we may notice late that IO was suspended, * and postpone, or schedule for retry, a write, * before it even was submitted or sent. * In that case we do not want to touch the bitmap at all. */ struct drbd_peer_device *peer_device = first_peer_device(device); if ((s & (RQ_POSTPONED|RQ_LOCAL_MASK|RQ_NET_MASK)) != RQ_POSTPONED) { if (!(s & RQ_NET_OK) || !(s & RQ_LOCAL_OK)) drbd_set_out_of_sync(peer_device, req->i.sector, req->i.size); if ((s & RQ_NET_OK) && (s & RQ_LOCAL_OK) && (s & RQ_NET_SIS)) drbd_set_in_sync(peer_device, req->i.sector, req->i.size); } /* one might be tempted to move the drbd_al_complete_io * to the local io completion callback drbd_request_endio. * but, if this was a mirror write, we may only * drbd_al_complete_io after this is RQ_NET_DONE, * otherwise the extent could be dropped from the al * before it has actually been written on the peer. * if we crash before our peer knows about the request, * but after the extent has been dropped from the al, * we would forget to resync the corresponding extent. */ if (s & RQ_IN_ACT_LOG) { if (get_ldev_if_state(device, D_FAILED)) { drbd_al_complete_io(device, &req->i); put_ldev(device); } else if (drbd_ratelimit()) { drbd_warn(device, "Should have called drbd_al_complete_io(, %llu, %u), " "but my Disk seems to have failed :(\n", (unsigned long long) req->i.sector, req->i.size); } } } mempool_free(req, &drbd_request_mempool);} static void wake_all_senders(struct drbd_connection *connection){ wake_up(&connection->sender_work.q_wait);} /* must hold resource->req_lock */void start_new_tl_epoch(struct drbd_connection *connection){ /* no point closing an epoch, if it is empty, anyways. */ if (connection->current_tle_writes == 0) return; connection->current_tle_writes = 0; atomic_inc(&connection->current_tle_nr); wake_all_senders(connection);} void complete_master_bio(struct drbd_device *device, struct bio_and_error *m){ if (unlikely(m->error)) m->bio->bi_status = errno_to_blk_status(m->error); bio_endio(m->bio); dec_ap_bio(device);} /* Helper for __req_mod(). * Set m->bio to the master bio, if it is fit to be completed, * or leave it alone (it is initialized to NULL in __req_mod), * if it has already been completed, or cannot be completed yet. * If m->bio is set, the error status to be returned is placed in m->error. */staticvoid drbd_req_complete(struct drbd_request *req, struct bio_and_error *m){ const unsigned s = req->rq_state; struct drbd_device *device = req->device; int error, ok; /* we must not complete the master bio, while it is * still being processed by _drbd_send_zc_bio (drbd_send_dblock) * not yet acknowledged by the peer * not yet completed by the local io subsystem * these flags may get cleared in any order by * the worker, * the receiver, * the bio_endio completion callbacks. */ if ((s & RQ_LOCAL_PENDING && !(s & RQ_LOCAL_ABORTED)) || (s & RQ_NET_QUEUED) || (s & RQ_NET_PENDING) || (s & RQ_COMPLETION_SUSP)) { drbd_err(device, "drbd_req_complete: Logic BUG rq_state = 0x%x\n", s); return; } if (!req->master_bio) { drbd_err(device, "drbd_req_complete: Logic BUG, master_bio == NULL!\n"); return; } /* * figure out whether to report success or failure. * * report success when at least one of the operations succeeded. * or, to put the other way, * only report failure, when both operations failed. * * what to do about the failures is handled elsewhere. * what we need to do here is just: complete the master_bio. * * local completion error, if any, has been stored as ERR_PTR * in private_bio within drbd_request_endio. */ ok = (s & RQ_LOCAL_OK) || (s & RQ_NET_OK); error = PTR_ERR(req->private_bio); /* Before we can signal completion to the upper layers, * we may need to close the current transfer log epoch. * We are within the request lock, so we can simply compare * the request epoch number with the current transfer log * epoch number. If they match, increase the current_tle_nr, * and reset the transfer log epoch write_cnt. */ if (op_is_write(bio_op(req->master_bio)) && req->epoch == atomic_read(&first_peer_device(device)->connection->current_tle_nr)) start_new_tl_epoch(first_peer_device(device)->connection); /* Update disk stats */ bio_end_io_acct(req->master_bio, req->start_jif); /* If READ failed, * have it be pushed back to the retry work queue, * so it will re-enter __drbd_make_request(), * and be re-assigned to a suitable local or remote path, * or failed if we do not have access to good data anymore. * * Unless it was failed early by __drbd_make_request(), * because no path was available, in which case * it was not even added to the transfer_log. * * read-ahead may fail, and will not be retried. * * WRITE should have used all available paths already. */ if (!ok && bio_op(req->master_bio) == REQ_OP_READ && !(req->master_bio->bi_opf & REQ_RAHEAD) && !list_empty(&req->tl_requests)) req->rq_state |= RQ_POSTPONED; if (!(req->rq_state & RQ_POSTPONED)) { m->error = ok ? 0 : (error ?: -EIO); m->bio = req->master_bio; req->master_bio = NULL; /* We leave it in the tree, to be able to verify later * write-acks in protocol != C during resync. * But we mark it as "complete", so it won't be counted as * conflict in a multi-primary setup. */ req->i.completed = true; } if (req->i.waiting) wake_up(&device->misc_wait); /* Either we are about to complete to upper layers, * or we will restart this request. * In either case, the request object will be destroyed soon, * so better remove it from all lists. */ list_del_init(&req->req_pending_master_completion);} /* still holds resource->req_lock */static void drbd_req_put_completion_ref(struct drbd_request *req, struct bio_and_error *m, int put){ struct drbd_device *device = req->device; D_ASSERT(device, m || (req->rq_state & RQ_POSTPONED)); if (!put) return; if (!atomic_sub_and_test(put, &req->completion_ref)) return; drbd_req_complete(req, m); /* local completion may still come in later, * we need to keep the req object around. */ if (req->rq_state & RQ_LOCAL_ABORTED) return; if (req->rq_state & RQ_POSTPONED) { /* don't destroy the req object just yet, * but queue it for retry */ drbd_restart_request(req); return; } kref_put(&req->kref, drbd_req_destroy);} static void set_if_null_req_next(struct drbd_peer_device *peer_device, struct drbd_request *req){ struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; if (!connection) return; if (connection->req_next == NULL) connection->req_next = req;} static void advance_conn_req_next(struct drbd_peer_device *peer_device, struct drbd_request *req){ struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; struct drbd_request *iter = req; if (!connection) return; if (connection->req_next != req) return; req = NULL; list_for_each_entry_continue(iter, &connection->transfer_log, tl_requests) { const unsigned int s = iter->rq_state; if (s & RQ_NET_QUEUED) { req = iter; break; } } connection->req_next = req;} static void set_if_null_req_ack_pending(struct drbd_peer_device *peer_device, struct drbd_request *req){ struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; if (!connection) return; if (connection->req_ack_pending == NULL) connection->req_ack_pending = req;} static void advance_conn_req_ack_pending(struct drbd_peer_device *peer_device, struct drbd_request *req){ struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; struct drbd_request *iter = req; if (!connection) return; if (connection->req_ack_pending != req) return; req = NULL; list_for_each_entry_continue(iter, &connection->transfer_log, tl_requests) { const unsigned int s = iter->rq_state; if ((s & RQ_NET_SENT) && (s & RQ_NET_PENDING)) { req = iter; break; } } connection->req_ack_pending = req;} static void set_if_null_req_not_net_done(struct drbd_peer_device *peer_device, struct drbd_request *req){ struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; if (!connection) return; if (connection->req_not_net_done == NULL) connection->req_not_net_done = req;} static void advance_conn_req_not_net_done(struct drbd_peer_device *peer_device, struct drbd_request *req){ struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; struct drbd_request *iter = req; if (!connection) return; if (connection->req_not_net_done != req) return; req = NULL; list_for_each_entry_continue(iter, &connection->transfer_log, tl_requests) { const unsigned int s = iter->rq_state; if ((s & RQ_NET_SENT) && !(s & RQ_NET_DONE)) { req = iter; break; } } connection->req_not_net_done = req;} /* I'd like this to be the only place that manipulates * req->completion_ref and req->kref. */static void mod_rq_state(struct drbd_request *req, struct bio_and_error *m, int clear, int set){ struct drbd_device *device = req->device; struct drbd_peer_device *peer_device = first_peer_device(device); unsigned s = req->rq_state; int c_put = 0; if (drbd_suspended(device) && !((s | clear) & RQ_COMPLETION_SUSP)) set |= RQ_COMPLETION_SUSP; /* apply */ req->rq_state &= ~clear; req->rq_state |= set; /* no change? */ if (req->rq_state == s) return; /* intent: get references */ kref_get(&req->kref); if (!(s & RQ_LOCAL_PENDING) && (set & RQ_LOCAL_PENDING)) atomic_inc(&req->completion_ref); if (!(s & RQ_NET_PENDING) && (set & RQ_NET_PENDING)) { inc_ap_pending(device); atomic_inc(&req->completion_ref); } if (!(s & RQ_NET_QUEUED) && (set & RQ_NET_QUEUED)) { atomic_inc(&req->completion_ref); set_if_null_req_next(peer_device, req); } if (!(s & RQ_EXP_BARR_ACK) && (set & RQ_EXP_BARR_ACK)) kref_get(&req->kref); /* wait for the DONE */ if (!(s & RQ_NET_SENT) && (set & RQ_NET_SENT)) { /* potentially already completed in the ack_receiver thread */ if (!(s & RQ_NET_DONE)) { atomic_add(req->i.size >> 9, &device->ap_in_flight); set_if_null_req_not_net_done(peer_device, req); } if (req->rq_state & RQ_NET_PENDING) set_if_null_req_ack_pending(peer_device, req); } if (!(s & RQ_COMPLETION_SUSP) && (set & RQ_COMPLETION_SUSP)) atomic_inc(&req->completion_ref); /* progress: put references */ if ((s & RQ_COMPLETION_SUSP) && (clear & RQ_COMPLETION_SUSP)) ++c_put; if (!(s & RQ_LOCAL_ABORTED) && (set & RQ_LOCAL_ABORTED)) { D_ASSERT(device, req->rq_state & RQ_LOCAL_PENDING); ++c_put; } if ((s & RQ_LOCAL_PENDING) && (clear & RQ_LOCAL_PENDING)) { if (req->rq_state & RQ_LOCAL_ABORTED) kref_put(&req->kref, drbd_req_destroy); else ++c_put; list_del_init(&req->req_pending_local); } if ((s & RQ_NET_PENDING) && (clear & RQ_NET_PENDING)) { dec_ap_pending(device); ++c_put; req->acked_jif = jiffies; advance_conn_req_ack_pending(peer_device, req); } if ((s & RQ_NET_QUEUED) && (clear & RQ_NET_QUEUED)) { ++c_put; advance_conn_req_next(peer_device, req); } if (!(s & RQ_NET_DONE) && (set & RQ_NET_DONE)) { if (s & RQ_NET_SENT) atomic_sub(req->i.size >> 9, &device->ap_in_flight); if (s & RQ_EXP_BARR_ACK) kref_put(&req->kref, drbd_req_destroy); req->net_done_jif = jiffies;Lines 1–500 of 1791. Load more
// SPDX-License-Identifier: GPL-2.0-only /* drbd_req.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/slab.h> #include <linux/drbd.h> #include "drbd_int.h" #include "drbd_req.h" static bool drbd_may_do_local_read(struct drbd_device *device, sector_t sector, int size); static struct drbd_request *drbd_req_new(struct drbd_device *device, struct bio *bio_src) { struct drbd_request *req; req = mempool_alloc(&drbd_request_mempool, GFP_NOIO); if (!req) return NULL; memset(req, 0, sizeof(*req)); req->rq_state = (bio_data_dir(bio_src) == WRITE ? RQ_WRITE : 0) | (bio_op(bio_src) == REQ_OP_WRITE_ZEROES ? RQ_ZEROES : 0) | (bio_op(bio_src) == REQ_OP_DISCARD ? RQ_UNMAP : 0); req->device = device; req->master_bio = bio_src; req->epoch = 0; drbd_clear_interval(&req->i); req->i.sector = bio_src->bi_iter.bi_sector; req->i.size = bio_src->bi_iter.bi_size; req->i.local = true; req->i.waiting = false; INIT_LIST_HEAD(&req->tl_requests); INIT_LIST_HEAD(&req->w.list); INIT_LIST_HEAD(&req->req_pending_master_completion); INIT_LIST_HEAD(&req->req_pending_local); /* one reference to be put by __drbd_make_request */ atomic_set(&req->completion_ref, 1); /* one kref as long as completion_ref > 0 */ kref_init(&req->kref); return req; } static void drbd_remove_request_interval(struct rb_root *root, struct drbd_request *req) { struct drbd_device *device = req->device; struct drbd_interval *i = &req->i; drbd_remove_interval(root, i); /* Wake up any processes waiting for this request to complete. */ if (i->waiting) wake_up(&device->misc_wait); } void drbd_req_destroy(struct kref *kref) { struct drbd_request *req = container_of(kref, struct drbd_request, kref); struct drbd_device *device = req->device; const unsigned s = req->rq_state; if ((req->master_bio && !(s & RQ_POSTPONED)) || atomic_read(&req->completion_ref) || (s & RQ_LOCAL_PENDING) || ((s & RQ_NET_MASK) && !(s & RQ_NET_DONE))) { drbd_err(device, "drbd_req_destroy: Logic BUG rq_state = 0x%x, completion_ref = %d\n", s, atomic_read(&req->completion_ref)); return; } /* If called from mod_rq_state (expected normal case) or * drbd_send_and_submit (the less likely normal path), this holds the * req_lock, and req->tl_requests will typicaly be on ->transfer_log, * though it may be still empty (never added to the transfer log). * * If called from do_retry(), we do NOT hold the req_lock, but we are * still allowed to unconditionally list_del(&req->tl_requests), * because it will be on a local on-stack list only. */ list_del_init(&req->tl_requests); /* finally remove the request from the conflict detection * respective block_id verification interval tree. */ if (!drbd_interval_empty(&req->i)) { struct rb_root *root; if (s & RQ_WRITE) root = &device->write_requests; else root = &device->read_requests; drbd_remove_request_interval(root, req); } else if (s & (RQ_NET_MASK & ~RQ_NET_DONE) && req->i.size != 0) drbd_err(device, "drbd_req_destroy: Logic BUG: interval empty, but: rq_state=0x%x, sect=%llu, size=%u\n", s, (unsigned long long)req->i.sector, req->i.size); /* if it was a write, we may have to set the corresponding * bit(s) out-of-sync first. If it had a local part, we need to * release the reference to the activity log. */ if (s & RQ_WRITE) { /* Set out-of-sync unless both OK flags are set * (local only or remote failed). * Other places where we set out-of-sync: * READ with local io-error */ /* There is a special case: * we may notice late that IO was suspended, * and postpone, or schedule for retry, a write, * before it even was submitted or sent. * In that case we do not want to touch the bitmap at all. */ struct drbd_peer_device *peer_device = first_peer_device(device); if ((s & (RQ_POSTPONED|RQ_LOCAL_MASK|RQ_NET_MASK)) != RQ_POSTPONED) { if (!(s & RQ_NET_OK) || !(s & RQ_LOCAL_OK)) drbd_set_out_of_sync(peer_device, req->i.sector, req->i.size); if ((s & RQ_NET_OK) && (s & RQ_LOCAL_OK) && (s & RQ_NET_SIS)) drbd_set_in_sync(peer_device, req->i.sector, req->i.size); } /* one might be tempted to move the drbd_al_complete_io * to the local io completion callback drbd_request_endio. * but, if this was a mirror write, we may only * drbd_al_complete_io after this is RQ_NET_DONE, * otherwise the extent could be dropped from the al * before it has actually been written on the peer. * if we crash before our peer knows about the request, * but after the extent has been dropped from the al, * we would forget to resync the corresponding extent. */ if (s & RQ_IN_ACT_LOG) { if (get_ldev_if_state(device, D_FAILED)) { drbd_al_complete_io(device, &req->i); put_ldev(device); } else if (drbd_ratelimit()) { drbd_warn(device, "Should have called drbd_al_complete_io(, %llu, %u), " "but my Disk seems to have failed :(\n", (unsigned long long) req->i.sector, req->i.size); } } } mempool_free(req, &drbd_request_mempool); } static void wake_all_senders(struct drbd_connection *connection) { wake_up(&connection->sender_work.q_wait); } /* must hold resource->req_lock */ void start_new_tl_epoch(struct drbd_connection *connection) { /* no point closing an epoch, if it is empty, anyways. */ if (connection->current_tle_writes == 0) return; connection->current_tle_writes = 0; atomic_inc(&connection->current_tle_nr); wake_all_senders(connection); } void complete_master_bio(struct drbd_device *device, struct bio_and_error *m) { if (unlikely(m->error)) m->bio->bi_status = errno_to_blk_status(m->error); bio_endio(m->bio); dec_ap_bio(device); } /* Helper for __req_mod(). * Set m->bio to the master bio, if it is fit to be completed, * or leave it alone (it is initialized to NULL in __req_mod), * if it has already been completed, or cannot be completed yet. * If m->bio is set, the error status to be returned is placed in m->error. */ static void drbd_req_complete(struct drbd_request *req, struct bio_and_error *m) { const unsigned s = req->rq_state; struct drbd_device *device = req->device; int error, ok; /* we must not complete the master bio, while it is * still being processed by _drbd_send_zc_bio (drbd_send_dblock) * not yet acknowledged by the peer * not yet completed by the local io subsystem * these flags may get cleared in any order by * the worker, * the receiver, * the bio_endio completion callbacks. */ if ((s & RQ_LOCAL_PENDING && !(s & RQ_LOCAL_ABORTED)) || (s & RQ_NET_QUEUED) || (s & RQ_NET_PENDING) || (s & RQ_COMPLETION_SUSP)) { drbd_err(device, "drbd_req_complete: Logic BUG rq_state = 0x%x\n", s); return; } if (!req->master_bio) { drbd_err(device, "drbd_req_complete: Logic BUG, master_bio == NULL!\n"); return; } /* * figure out whether to report success or failure. * * report success when at least one of the operations succeeded. * or, to put the other way, * only report failure, when both operations failed. * * what to do about the failures is handled elsewhere. * what we need to do here is just: complete the master_bio. * * local completion error, if any, has been stored as ERR_PTR * in private_bio within drbd_request_endio. */ ok = (s & RQ_LOCAL_OK) || (s & RQ_NET_OK); error = PTR_ERR(req->private_bio); /* Before we can signal completion to the upper layers, * we may need to close the current transfer log epoch. * We are within the request lock, so we can simply compare * the request epoch number with the current transfer log * epoch number. If they match, increase the current_tle_nr, * and reset the transfer log epoch write_cnt. */ if (op_is_write(bio_op(req->master_bio)) && req->epoch == atomic_read(&first_peer_device(device)->connection->current_tle_nr)) start_new_tl_epoch(first_peer_device(device)->connection); /* Update disk stats */ bio_end_io_acct(req->master_bio, req->start_jif); /* If READ failed, * have it be pushed back to the retry work queue, * so it will re-enter __drbd_make_request(), * and be re-assigned to a suitable local or remote path, * or failed if we do not have access to good data anymore. * * Unless it was failed early by __drbd_make_request(), * because no path was available, in which case * it was not even added to the transfer_log. * * read-ahead may fail, and will not be retried. * * WRITE should have used all available paths already. */ if (!ok && bio_op(req->master_bio) == REQ_OP_READ && !(req->master_bio->bi_opf & REQ_RAHEAD) && !list_empty(&req->tl_requests)) req->rq_state |= RQ_POSTPONED; if (!(req->rq_state & RQ_POSTPONED)) { m->error = ok ? 0 : (error ?: -EIO); m->bio = req->master_bio; req->master_bio = NULL; /* We leave it in the tree, to be able to verify later * write-acks in protocol != C during resync. * But we mark it as "complete", so it won't be counted as * conflict in a multi-primary setup. */ req->i.completed = true; } if (req->i.waiting) wake_up(&device->misc_wait); /* Either we are about to complete to upper layers, * or we will restart this request. * In either case, the request object will be destroyed soon, * so better remove it from all lists. */ list_del_init(&req->req_pending_master_completion); } /* still holds resource->req_lock */ static void drbd_req_put_completion_ref(struct drbd_request *req, struct bio_and_error *m, int put) { struct drbd_device *device = req->device; D_ASSERT(device, m || (req->rq_state & RQ_POSTPONED)); if (!put) return; if (!atomic_sub_and_test(put, &req->completion_ref)) return; drbd_req_complete(req, m); /* local completion may still come in later, * we need to keep the req object around. */ if (req->rq_state & RQ_LOCAL_ABORTED) return; if (req->rq_state & RQ_POSTPONED) { /* don't destroy the req object just yet, * but queue it for retry */ drbd_restart_request(req); return; } kref_put(&req->kref, drbd_req_destroy); } static void set_if_null_req_next(struct drbd_peer_device *peer_device, struct drbd_request *req) { struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; if (!connection) return; if (connection->req_next == NULL) connection->req_next = req; } static void advance_conn_req_next(struct drbd_peer_device *peer_device, struct drbd_request *req) { struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; struct drbd_request *iter = req; if (!connection) return; if (connection->req_next != req) return; req = NULL; list_for_each_entry_continue(iter, &connection->transfer_log, tl_requests) { const unsigned int s = iter->rq_state; if (s & RQ_NET_QUEUED) { req = iter; break; } } connection->req_next = req; } static void set_if_null_req_ack_pending(struct drbd_peer_device *peer_device, struct drbd_request *req) { struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; if (!connection) return; if (connection->req_ack_pending == NULL) connection->req_ack_pending = req; } static void advance_conn_req_ack_pending(struct drbd_peer_device *peer_device, struct drbd_request *req) { struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; struct drbd_request *iter = req; if (!connection) return; if (connection->req_ack_pending != req) return; req = NULL; list_for_each_entry_continue(iter, &connection->transfer_log, tl_requests) { const unsigned int s = iter->rq_state; if ((s & RQ_NET_SENT) && (s & RQ_NET_PENDING)) { req = iter; break; } } connection->req_ack_pending = req; } static void set_if_null_req_not_net_done(struct drbd_peer_device *peer_device, struct drbd_request *req) { struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; if (!connection) return; if (connection->req_not_net_done == NULL) connection->req_not_net_done = req; } static void advance_conn_req_not_net_done(struct drbd_peer_device *peer_device, struct drbd_request *req) { struct drbd_connection *connection = peer_device ? peer_device->connection : NULL; struct drbd_request *iter = req; if (!connection) return; if (connection->req_not_net_done != req) return; req = NULL; list_for_each_entry_continue(iter, &connection->transfer_log, tl_requests) { const unsigned int s = iter->rq_state; if ((s & RQ_NET_SENT) && !(s & RQ_NET_DONE)) { req = iter; break; } } connection->req_not_net_done = req; } /* I'd like this to be the only place that manipulates * req->completion_ref and req->kref. */ static void mod_rq_state(struct drbd_request *req, struct bio_and_error *m, int clear, int set) { struct drbd_device *device = req->device; struct drbd_peer_device *peer_device = first_peer_device(device); unsigned s = req->rq_state; int c_put = 0; if (drbd_suspended(device) && !((s | clear) & RQ_COMPLETION_SUSP)) set |= RQ_COMPLETION_SUSP; /* apply */ req->rq_state &= ~clear; req->rq_state |= set; /* no change? */ if (req->rq_state == s) return; /* intent: get references */ kref_get(&req->kref); if (!(s & RQ_LOCAL_PENDING) && (set & RQ_LOCAL_PENDING)) atomic_inc(&req->completion_ref); if (!(s & RQ_NET_PENDING) && (set & RQ_NET_PENDING)) { inc_ap_pending(device); atomic_inc(&req->completion_ref); } if (!(s & RQ_NET_QUEUED) && (set & RQ_NET_QUEUED)) { atomic_inc(&req->completion_ref); set_if_null_req_next(peer_device, req); } if (!(s & RQ_EXP_BARR_ACK) && (set & RQ_EXP_BARR_ACK)) kref_get(&req->kref); /* wait for the DONE */ if (!(s & RQ_NET_SENT) && (set & RQ_NET_SENT)) { /* potentially already completed in the ack_receiver thread */ if (!(s & RQ_NET_DONE)) { atomic_add(req->i.size >> 9, &device->ap_in_flight); set_if_null_req_not_net_done(peer_device, req); } if (req->rq_state & RQ_NET_PENDING) set_if_null_req_ack_pending(peer_device, req); } if (!(s & RQ_COMPLETION_SUSP) && (set & RQ_COMPLETION_SUSP)) atomic_inc(&req->completion_ref); /* progress: put references */ if ((s & RQ_COMPLETION_SUSP) && (clear & RQ_COMPLETION_SUSP)) ++c_put; if (!(s & RQ_LOCAL_ABORTED) && (set & RQ_LOCAL_ABORTED)) { D_ASSERT(device, req->rq_state & RQ_LOCAL_PENDING); ++c_put; } if ((s & RQ_LOCAL_PENDING) && (clear & RQ_LOCAL_PENDING)) { if (req->rq_state & RQ_LOCAL_ABORTED) kref_put(&req->kref, drbd_req_destroy); else ++c_put; list_del_init(&req->req_pending_local); } if ((s & RQ_NET_PENDING) && (clear & RQ_NET_PENDING)) { dec_ap_pending(device); ++c_put; req->acked_jif = jiffies; advance_conn_req_ack_pending(peer_device, req); } if ((s & RQ_NET_QUEUED) && (clear & RQ_NET_QUEUED)) { ++c_put; advance_conn_req_next(peer_device, req); } if (!(s & RQ_NET_DONE) && (set & RQ_NET_DONE)) { if (s & RQ_NET_SENT) atomic_sub(req->i.size >> 9, &device->ap_in_flight); if (s & RQ_EXP_BARR_ACK) kref_put(&req->kref, drbd_req_destroy); req->net_done_jif = jiffies;