Back to home page

OSCL-LXR
 
 

     

0001 // SPDX-License-Identifier: GPL-2.0-or-later
0002 /*
0003  * RDMA Network Block Driver
0004  *
0005  * Copyright (c) 2014 - 2018 ProfitBricks GmbH. All rights reserved.
0006  * Copyright (c) 2018 - 2019 1&1 IONOS Cloud GmbH. All rights reserved.
0007  * Copyright (c) 2019 - 2020 1&1 IONOS SE. All rights reserved.
0008  */
0009 
0010 #undef pr_fmt
0011 #define pr_fmt(fmt) KBUILD_MODNAME " L" __stringify(__LINE__) ": " fmt
0012 
0013 #include <linux/module.h>
0014 #include <linux/blkdev.h>
0015 #include <linux/hdreg.h>
0016 #include <linux/scatterlist.h>
0017 #include <linux/idr.h>
0018 
0019 #include "rnbd-clt.h"
0020 
0021 MODULE_DESCRIPTION("RDMA Network Block Device Client");
0022 MODULE_LICENSE("GPL");
0023 
0024 static int rnbd_client_major;
0025 static DEFINE_IDA(index_ida);
0026 static DEFINE_MUTEX(sess_lock);
0027 static LIST_HEAD(sess_list);
0028 static struct workqueue_struct *rnbd_clt_wq;
0029 
0030 /*
0031  * Maximum number of partitions an instance can have.
0032  * 6 bits = 64 minors = 63 partitions (one minor is used for the device itself)
0033  */
0034 #define RNBD_PART_BITS      6
0035 
0036 static inline bool rnbd_clt_get_sess(struct rnbd_clt_session *sess)
0037 {
0038     return refcount_inc_not_zero(&sess->refcount);
0039 }
0040 
0041 static void free_sess(struct rnbd_clt_session *sess);
0042 
0043 static void rnbd_clt_put_sess(struct rnbd_clt_session *sess)
0044 {
0045     might_sleep();
0046 
0047     if (refcount_dec_and_test(&sess->refcount))
0048         free_sess(sess);
0049 }
0050 
0051 static void rnbd_clt_put_dev(struct rnbd_clt_dev *dev)
0052 {
0053     might_sleep();
0054 
0055     if (!refcount_dec_and_test(&dev->refcount))
0056         return;
0057 
0058     ida_free(&index_ida, dev->clt_device_id);
0059     kfree(dev->hw_queues);
0060     kfree(dev->pathname);
0061     rnbd_clt_put_sess(dev->sess);
0062     mutex_destroy(&dev->lock);
0063     kfree(dev);
0064 }
0065 
0066 static inline bool rnbd_clt_get_dev(struct rnbd_clt_dev *dev)
0067 {
0068     return refcount_inc_not_zero(&dev->refcount);
0069 }
0070 
0071 static void rnbd_clt_change_capacity(struct rnbd_clt_dev *dev,
0072                     sector_t new_nsectors)
0073 {
0074     if (get_capacity(dev->gd) == new_nsectors)
0075         return;
0076 
0077     /*
0078      * If the size changed, we need to revalidate it
0079      */
0080     rnbd_clt_info(dev, "Device size changed from %llu to %llu sectors\n",
0081               get_capacity(dev->gd), new_nsectors);
0082     set_capacity_and_notify(dev->gd, new_nsectors);
0083 }
0084 
0085 static int process_msg_open_rsp(struct rnbd_clt_dev *dev,
0086                 struct rnbd_msg_open_rsp *rsp)
0087 {
0088     struct kobject *gd_kobj;
0089     int err = 0;
0090 
0091     mutex_lock(&dev->lock);
0092     if (dev->dev_state == DEV_STATE_UNMAPPED) {
0093         rnbd_clt_info(dev,
0094                    "Ignoring Open-Response message from server for  unmapped device\n");
0095         err = -ENOENT;
0096         goto out;
0097     }
0098     if (dev->dev_state == DEV_STATE_MAPPED_DISCONNECTED) {
0099         u64 nsectors = le64_to_cpu(rsp->nsectors);
0100 
0101         rnbd_clt_change_capacity(dev, nsectors);
0102         gd_kobj = &disk_to_dev(dev->gd)->kobj;
0103         kobject_uevent(gd_kobj, KOBJ_ONLINE);
0104         rnbd_clt_info(dev, "Device online, device remapped successfully\n");
0105     }
0106     if (!rsp->logical_block_size) {
0107         err = -EINVAL;
0108         goto out;
0109     }
0110     dev->device_id = le32_to_cpu(rsp->device_id);
0111     dev->dev_state = DEV_STATE_MAPPED;
0112 
0113 out:
0114     mutex_unlock(&dev->lock);
0115 
0116     return err;
0117 }
0118 
0119 int rnbd_clt_resize_disk(struct rnbd_clt_dev *dev, sector_t newsize)
0120 {
0121     int ret = 0;
0122 
0123     mutex_lock(&dev->lock);
0124     if (dev->dev_state != DEV_STATE_MAPPED) {
0125         pr_err("Failed to set new size of the device, device is not opened\n");
0126         ret = -ENOENT;
0127         goto out;
0128     }
0129     rnbd_clt_change_capacity(dev, newsize);
0130 
0131 out:
0132     mutex_unlock(&dev->lock);
0133 
0134     return ret;
0135 }
0136 
0137 static inline void rnbd_clt_dev_requeue(struct rnbd_queue *q)
0138 {
0139     if (WARN_ON(!q->hctx))
0140         return;
0141 
0142     /* We can come here from interrupt, thus async=true */
0143     blk_mq_run_hw_queue(q->hctx, true);
0144 }
0145 
0146 enum {
0147     RNBD_DELAY_IFBUSY = -1,
0148 };
0149 
0150 /**
0151  * rnbd_get_cpu_qlist() - finds a list with HW queues to be rerun
0152  * @sess:   Session to find a queue for
0153  * @cpu:    Cpu to start the search from
0154  *
0155  * Description:
0156  *     Each CPU has a list of HW queues, which needs to be rerun.  If a list
0157  *     is not empty - it is marked with a bit.  This function finds first
0158  *     set bit in a bitmap and returns corresponding CPU list.
0159  */
0160 static struct rnbd_cpu_qlist *
0161 rnbd_get_cpu_qlist(struct rnbd_clt_session *sess, int cpu)
0162 {
0163     int bit;
0164 
0165     /* Search from cpu to nr_cpu_ids */
0166     bit = find_next_bit(sess->cpu_queues_bm, nr_cpu_ids, cpu);
0167     if (bit < nr_cpu_ids) {
0168         return per_cpu_ptr(sess->cpu_queues, bit);
0169     } else if (cpu != 0) {
0170         /* Search from 0 to cpu */
0171         bit = find_first_bit(sess->cpu_queues_bm, cpu);
0172         if (bit < cpu)
0173             return per_cpu_ptr(sess->cpu_queues, bit);
0174     }
0175 
0176     return NULL;
0177 }
0178 
0179 static inline int nxt_cpu(int cpu)
0180 {
0181     return (cpu + 1) % nr_cpu_ids;
0182 }
0183 
0184 /**
0185  * rnbd_rerun_if_needed() - rerun next queue marked as stopped
0186  * @sess:   Session to rerun a queue on
0187  *
0188  * Description:
0189  *     Each CPU has it's own list of HW queues, which should be rerun.
0190  *     Function finds such list with HW queues, takes a list lock, picks up
0191  *     the first HW queue out of the list and requeues it.
0192  *
0193  * Return:
0194  *     True if the queue was requeued, false otherwise.
0195  *
0196  * Context:
0197  *     Does not matter.
0198  */
0199 static bool rnbd_rerun_if_needed(struct rnbd_clt_session *sess)
0200 {
0201     struct rnbd_queue *q = NULL;
0202     struct rnbd_cpu_qlist *cpu_q;
0203     unsigned long flags;
0204     int *cpup;
0205 
0206     /*
0207      * To keep fairness and not to let other queues starve we always
0208      * try to wake up someone else in round-robin manner.  That of course
0209      * increases latency but queues always have a chance to be executed.
0210      */
0211     cpup = get_cpu_ptr(sess->cpu_rr);
0212     for (cpu_q = rnbd_get_cpu_qlist(sess, nxt_cpu(*cpup)); cpu_q;
0213          cpu_q = rnbd_get_cpu_qlist(sess, nxt_cpu(cpu_q->cpu))) {
0214         if (!spin_trylock_irqsave(&cpu_q->requeue_lock, flags))
0215             continue;
0216         if (!test_bit(cpu_q->cpu, sess->cpu_queues_bm))
0217             goto unlock;
0218         q = list_first_entry_or_null(&cpu_q->requeue_list,
0219                          typeof(*q), requeue_list);
0220         if (WARN_ON(!q))
0221             goto clear_bit;
0222         list_del_init(&q->requeue_list);
0223         clear_bit_unlock(0, &q->in_list);
0224 
0225         if (list_empty(&cpu_q->requeue_list)) {
0226             /* Clear bit if nothing is left */
0227 clear_bit:
0228             clear_bit(cpu_q->cpu, sess->cpu_queues_bm);
0229         }
0230 unlock:
0231         spin_unlock_irqrestore(&cpu_q->requeue_lock, flags);
0232 
0233         if (q)
0234             break;
0235     }
0236 
0237     /**
0238      * Saves the CPU that is going to be requeued on the per-cpu var. Just
0239      * incrementing it doesn't work because rnbd_get_cpu_qlist() will
0240      * always return the first CPU with something on the queue list when the
0241      * value stored on the var is greater than the last CPU with something
0242      * on the list.
0243      */
0244     if (cpu_q)
0245         *cpup = cpu_q->cpu;
0246     put_cpu_ptr(sess->cpu_rr);
0247 
0248     if (q)
0249         rnbd_clt_dev_requeue(q);
0250 
0251     return q;
0252 }
0253 
0254 /**
0255  * rnbd_rerun_all_if_idle() - rerun all queues left in the list if
0256  *               session is idling (there are no requests
0257  *               in-flight).
0258  * @sess:   Session to rerun the queues on
0259  *
0260  * Description:
0261  *     This function tries to rerun all stopped queues if there are no
0262  *     requests in-flight anymore.  This function tries to solve an obvious
0263  *     problem, when number of tags < than number of queues (hctx), which
0264  *     are stopped and put to sleep.  If last permit, which has been just put,
0265  *     does not wake up all left queues (hctxs), IO requests hang forever.
0266  *
0267  *     That can happen when all number of permits, say N, have been exhausted
0268  *     from one CPU, and we have many block devices per session, say M.
0269  *     Each block device has it's own queue (hctx) for each CPU, so eventually
0270  *     we can put that number of queues (hctxs) to sleep: M x nr_cpu_ids.
0271  *     If number of permits N < M x nr_cpu_ids finally we will get an IO hang.
0272  *
0273  *     To avoid this hang last caller of rnbd_put_permit() (last caller is the
0274  *     one who observes sess->busy == 0) must wake up all remaining queues.
0275  *
0276  * Context:
0277  *     Does not matter.
0278  */
0279 static void rnbd_rerun_all_if_idle(struct rnbd_clt_session *sess)
0280 {
0281     bool requeued;
0282 
0283     do {
0284         requeued = rnbd_rerun_if_needed(sess);
0285     } while (atomic_read(&sess->busy) == 0 && requeued);
0286 }
0287 
0288 static struct rtrs_permit *rnbd_get_permit(struct rnbd_clt_session *sess,
0289                          enum rtrs_clt_con_type con_type,
0290                          enum wait_type wait)
0291 {
0292     struct rtrs_permit *permit;
0293 
0294     permit = rtrs_clt_get_permit(sess->rtrs, con_type, wait);
0295     if (permit)
0296         /* We have a subtle rare case here, when all permits can be
0297          * consumed before busy counter increased.  This is safe,
0298          * because loser will get NULL as a permit, observe 0 busy
0299          * counter and immediately restart the queue himself.
0300          */
0301         atomic_inc(&sess->busy);
0302 
0303     return permit;
0304 }
0305 
0306 static void rnbd_put_permit(struct rnbd_clt_session *sess,
0307                  struct rtrs_permit *permit)
0308 {
0309     rtrs_clt_put_permit(sess->rtrs, permit);
0310     atomic_dec(&sess->busy);
0311     /* Paired with rnbd_clt_dev_add_to_requeue().  Decrement first
0312      * and then check queue bits.
0313      */
0314     smp_mb__after_atomic();
0315     rnbd_rerun_all_if_idle(sess);
0316 }
0317 
0318 static struct rnbd_iu *rnbd_get_iu(struct rnbd_clt_session *sess,
0319                      enum rtrs_clt_con_type con_type,
0320                      enum wait_type wait)
0321 {
0322     struct rnbd_iu *iu;
0323     struct rtrs_permit *permit;
0324 
0325     iu = kzalloc(sizeof(*iu), GFP_KERNEL);
0326     if (!iu)
0327         return NULL;
0328 
0329     permit = rnbd_get_permit(sess, con_type, wait);
0330     if (!permit) {
0331         kfree(iu);
0332         return NULL;
0333     }
0334 
0335     iu->permit = permit;
0336     /*
0337      * 1st reference is dropped after finishing sending a "user" message,
0338      * 2nd reference is dropped after confirmation with the response is
0339      * returned.
0340      * 1st and 2nd can happen in any order, so the rnbd_iu should be
0341      * released (rtrs_permit returned to rtrs) only after both
0342      * are finished.
0343      */
0344     atomic_set(&iu->refcount, 2);
0345     init_waitqueue_head(&iu->comp.wait);
0346     iu->comp.errno = INT_MAX;
0347 
0348     if (sg_alloc_table(&iu->sgt, 1, GFP_KERNEL)) {
0349         rnbd_put_permit(sess, permit);
0350         kfree(iu);
0351         return NULL;
0352     }
0353 
0354     return iu;
0355 }
0356 
0357 static void rnbd_put_iu(struct rnbd_clt_session *sess, struct rnbd_iu *iu)
0358 {
0359     if (atomic_dec_and_test(&iu->refcount)) {
0360         sg_free_table(&iu->sgt);
0361         rnbd_put_permit(sess, iu->permit);
0362         kfree(iu);
0363     }
0364 }
0365 
0366 static void rnbd_softirq_done_fn(struct request *rq)
0367 {
0368     struct rnbd_clt_dev *dev    = rq->q->disk->private_data;
0369     struct rnbd_clt_session *sess   = dev->sess;
0370     struct rnbd_iu *iu;
0371 
0372     iu = blk_mq_rq_to_pdu(rq);
0373     sg_free_table_chained(&iu->sgt, RNBD_INLINE_SG_CNT);
0374     rnbd_put_permit(sess, iu->permit);
0375     blk_mq_end_request(rq, errno_to_blk_status(iu->errno));
0376 }
0377 
0378 static void msg_io_conf(void *priv, int errno)
0379 {
0380     struct rnbd_iu *iu = priv;
0381     struct rnbd_clt_dev *dev = iu->dev;
0382     struct request *rq = iu->rq;
0383     int rw = rq_data_dir(rq);
0384 
0385     iu->errno = errno;
0386 
0387     blk_mq_complete_request(rq);
0388 
0389     if (errno)
0390         rnbd_clt_info_rl(dev, "%s I/O failed with err: %d\n",
0391                  rw == READ ? "read" : "write", errno);
0392 }
0393 
0394 static void wake_up_iu_comp(struct rnbd_iu *iu, int errno)
0395 {
0396     iu->comp.errno = errno;
0397     wake_up(&iu->comp.wait);
0398 }
0399 
0400 static void msg_conf(void *priv, int errno)
0401 {
0402     struct rnbd_iu *iu = priv;
0403 
0404     iu->errno = errno;
0405     schedule_work(&iu->work);
0406 }
0407 
0408 static int send_usr_msg(struct rtrs_clt_sess *rtrs, int dir,
0409             struct rnbd_iu *iu, struct kvec *vec,
0410             size_t len, struct scatterlist *sg, unsigned int sg_len,
0411             void (*conf)(struct work_struct *work),
0412             int *errno, int wait)
0413 {
0414     int err;
0415     struct rtrs_clt_req_ops req_ops;
0416 
0417     INIT_WORK(&iu->work, conf);
0418     req_ops = (struct rtrs_clt_req_ops) {
0419         .priv = iu,
0420         .conf_fn = msg_conf,
0421     };
0422     err = rtrs_clt_request(dir, &req_ops, rtrs, iu->permit,
0423                 vec, 1, len, sg, sg_len);
0424     if (!err && wait) {
0425         wait_event(iu->comp.wait, iu->comp.errno != INT_MAX);
0426         *errno = iu->comp.errno;
0427     } else {
0428         *errno = 0;
0429     }
0430 
0431     return err;
0432 }
0433 
0434 static void msg_close_conf(struct work_struct *work)
0435 {
0436     struct rnbd_iu *iu = container_of(work, struct rnbd_iu, work);
0437     struct rnbd_clt_dev *dev = iu->dev;
0438 
0439     wake_up_iu_comp(iu, iu->errno);
0440     rnbd_put_iu(dev->sess, iu);
0441     rnbd_clt_put_dev(dev);
0442 }
0443 
0444 static int send_msg_close(struct rnbd_clt_dev *dev, u32 device_id,
0445               enum wait_type wait)
0446 {
0447     struct rnbd_clt_session *sess = dev->sess;
0448     struct rnbd_msg_close msg;
0449     struct rnbd_iu *iu;
0450     struct kvec vec = {
0451         .iov_base = &msg,
0452         .iov_len  = sizeof(msg)
0453     };
0454     int err, errno;
0455 
0456     iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
0457     if (!iu)
0458         return -ENOMEM;
0459 
0460     iu->buf = NULL;
0461     iu->dev = dev;
0462 
0463     msg.hdr.type    = cpu_to_le16(RNBD_MSG_CLOSE);
0464     msg.device_id   = cpu_to_le32(device_id);
0465 
0466     WARN_ON(!rnbd_clt_get_dev(dev));
0467     err = send_usr_msg(sess->rtrs, WRITE, iu, &vec, 0, NULL, 0,
0468                msg_close_conf, &errno, wait);
0469     if (err) {
0470         rnbd_clt_put_dev(dev);
0471         rnbd_put_iu(sess, iu);
0472     } else {
0473         err = errno;
0474     }
0475 
0476     rnbd_put_iu(sess, iu);
0477     return err;
0478 }
0479 
0480 static void msg_open_conf(struct work_struct *work)
0481 {
0482     struct rnbd_iu *iu = container_of(work, struct rnbd_iu, work);
0483     struct rnbd_msg_open_rsp *rsp = iu->buf;
0484     struct rnbd_clt_dev *dev = iu->dev;
0485     int errno = iu->errno;
0486     bool from_map = false;
0487 
0488     /* INIT state is only triggered from rnbd_clt_map_device */
0489     if (dev->dev_state == DEV_STATE_INIT)
0490         from_map = true;
0491 
0492     if (errno) {
0493         rnbd_clt_err(dev,
0494                   "Opening failed, server responded: %d\n",
0495                   errno);
0496     } else {
0497         errno = process_msg_open_rsp(dev, rsp);
0498         if (errno) {
0499             u32 device_id = le32_to_cpu(rsp->device_id);
0500             /*
0501              * If server thinks its fine, but we fail to process
0502              * then be nice and send a close to server.
0503              */
0504             send_msg_close(dev, device_id, RTRS_PERMIT_NOWAIT);
0505         }
0506     }
0507     /* We free rsp in rnbd_clt_map_device for map scenario */
0508     if (!from_map)
0509         kfree(rsp);
0510     wake_up_iu_comp(iu, errno);
0511     rnbd_put_iu(dev->sess, iu);
0512     rnbd_clt_put_dev(dev);
0513 }
0514 
0515 static void msg_sess_info_conf(struct work_struct *work)
0516 {
0517     struct rnbd_iu *iu = container_of(work, struct rnbd_iu, work);
0518     struct rnbd_msg_sess_info_rsp *rsp = iu->buf;
0519     struct rnbd_clt_session *sess = iu->sess;
0520 
0521     if (!iu->errno)
0522         sess->ver = min_t(u8, rsp->ver, RNBD_PROTO_VER_MAJOR);
0523 
0524     kfree(rsp);
0525     wake_up_iu_comp(iu, iu->errno);
0526     rnbd_put_iu(sess, iu);
0527     rnbd_clt_put_sess(sess);
0528 }
0529 
0530 static int send_msg_open(struct rnbd_clt_dev *dev, enum wait_type wait)
0531 {
0532     struct rnbd_clt_session *sess = dev->sess;
0533     struct rnbd_msg_open_rsp *rsp;
0534     struct rnbd_msg_open msg;
0535     struct rnbd_iu *iu;
0536     struct kvec vec = {
0537         .iov_base = &msg,
0538         .iov_len  = sizeof(msg)
0539     };
0540     int err, errno;
0541 
0542     rsp = kzalloc(sizeof(*rsp), GFP_KERNEL);
0543     if (!rsp)
0544         return -ENOMEM;
0545 
0546     iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
0547     if (!iu) {
0548         kfree(rsp);
0549         return -ENOMEM;
0550     }
0551 
0552     iu->buf = rsp;
0553     iu->dev = dev;
0554 
0555     sg_init_one(iu->sgt.sgl, rsp, sizeof(*rsp));
0556 
0557     msg.hdr.type    = cpu_to_le16(RNBD_MSG_OPEN);
0558     msg.access_mode = dev->access_mode;
0559     strscpy(msg.dev_name, dev->pathname, sizeof(msg.dev_name));
0560 
0561     WARN_ON(!rnbd_clt_get_dev(dev));
0562     err = send_usr_msg(sess->rtrs, READ, iu,
0563                &vec, sizeof(*rsp), iu->sgt.sgl, 1,
0564                msg_open_conf, &errno, wait);
0565     if (err) {
0566         rnbd_clt_put_dev(dev);
0567         rnbd_put_iu(sess, iu);
0568         kfree(rsp);
0569     } else {
0570         err = errno;
0571     }
0572 
0573     rnbd_put_iu(sess, iu);
0574     return err;
0575 }
0576 
0577 static int send_msg_sess_info(struct rnbd_clt_session *sess, enum wait_type wait)
0578 {
0579     struct rnbd_msg_sess_info_rsp *rsp;
0580     struct rnbd_msg_sess_info msg;
0581     struct rnbd_iu *iu;
0582     struct kvec vec = {
0583         .iov_base = &msg,
0584         .iov_len  = sizeof(msg)
0585     };
0586     int err, errno;
0587 
0588     rsp = kzalloc(sizeof(*rsp), GFP_KERNEL);
0589     if (!rsp)
0590         return -ENOMEM;
0591 
0592     iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
0593     if (!iu) {
0594         kfree(rsp);
0595         return -ENOMEM;
0596     }
0597 
0598     iu->buf = rsp;
0599     iu->sess = sess;
0600     sg_init_one(iu->sgt.sgl, rsp, sizeof(*rsp));
0601 
0602     msg.hdr.type = cpu_to_le16(RNBD_MSG_SESS_INFO);
0603     msg.ver      = RNBD_PROTO_VER_MAJOR;
0604 
0605     if (!rnbd_clt_get_sess(sess)) {
0606         /*
0607          * That can happen only in one case, when RTRS has restablished
0608          * the connection and link_ev() is called, but session is almost
0609          * dead, last reference on session is put and caller is waiting
0610          * for RTRS to close everything.
0611          */
0612         err = -ENODEV;
0613         goto put_iu;
0614     }
0615     err = send_usr_msg(sess->rtrs, READ, iu,
0616                &vec, sizeof(*rsp), iu->sgt.sgl, 1,
0617                msg_sess_info_conf, &errno, wait);
0618     if (err) {
0619         rnbd_clt_put_sess(sess);
0620 put_iu:
0621         rnbd_put_iu(sess, iu);
0622         kfree(rsp);
0623     } else {
0624         err = errno;
0625     }
0626     rnbd_put_iu(sess, iu);
0627     return err;
0628 }
0629 
0630 static void set_dev_states_to_disconnected(struct rnbd_clt_session *sess)
0631 {
0632     struct rnbd_clt_dev *dev;
0633     struct kobject *gd_kobj;
0634 
0635     mutex_lock(&sess->lock);
0636     list_for_each_entry(dev, &sess->devs_list, list) {
0637         rnbd_clt_err(dev, "Device disconnected.\n");
0638 
0639         mutex_lock(&dev->lock);
0640         if (dev->dev_state == DEV_STATE_MAPPED) {
0641             dev->dev_state = DEV_STATE_MAPPED_DISCONNECTED;
0642             gd_kobj = &disk_to_dev(dev->gd)->kobj;
0643             kobject_uevent(gd_kobj, KOBJ_OFFLINE);
0644         }
0645         mutex_unlock(&dev->lock);
0646     }
0647     mutex_unlock(&sess->lock);
0648 }
0649 
0650 static void remap_devs(struct rnbd_clt_session *sess)
0651 {
0652     struct rnbd_clt_dev *dev;
0653     struct rtrs_attrs attrs;
0654     int err;
0655 
0656     /*
0657      * Careful here: we are called from RTRS link event directly,
0658      * thus we can't send any RTRS request and wait for response
0659      * or RTRS will not be able to complete request with failure
0660      * if something goes wrong (failing of outstanding requests
0661      * happens exactly from the context where we are blocking now).
0662      *
0663      * So to avoid deadlocks each usr message sent from here must
0664      * be asynchronous.
0665      */
0666 
0667     err = send_msg_sess_info(sess, RTRS_PERMIT_NOWAIT);
0668     if (err) {
0669         pr_err("send_msg_sess_info(\"%s\"): %d\n", sess->sessname, err);
0670         return;
0671     }
0672 
0673     err = rtrs_clt_query(sess->rtrs, &attrs);
0674     if (err) {
0675         pr_err("rtrs_clt_query(\"%s\"): %d\n", sess->sessname, err);
0676         return;
0677     }
0678     mutex_lock(&sess->lock);
0679     sess->max_io_size = attrs.max_io_size;
0680 
0681     list_for_each_entry(dev, &sess->devs_list, list) {
0682         bool skip;
0683 
0684         mutex_lock(&dev->lock);
0685         skip = (dev->dev_state == DEV_STATE_INIT);
0686         mutex_unlock(&dev->lock);
0687         if (skip)
0688             /*
0689              * When device is establishing connection for the first
0690              * time - do not remap, it will be closed soon.
0691              */
0692             continue;
0693 
0694         rnbd_clt_info(dev, "session reconnected, remapping device\n");
0695         err = send_msg_open(dev, RTRS_PERMIT_NOWAIT);
0696         if (err) {
0697             rnbd_clt_err(dev, "send_msg_open(): %d\n", err);
0698             break;
0699         }
0700     }
0701     mutex_unlock(&sess->lock);
0702 }
0703 
0704 static void rnbd_clt_link_ev(void *priv, enum rtrs_clt_link_ev ev)
0705 {
0706     struct rnbd_clt_session *sess = priv;
0707 
0708     switch (ev) {
0709     case RTRS_CLT_LINK_EV_DISCONNECTED:
0710         set_dev_states_to_disconnected(sess);
0711         break;
0712     case RTRS_CLT_LINK_EV_RECONNECTED:
0713         remap_devs(sess);
0714         break;
0715     default:
0716         pr_err("Unknown session event received (%d), session: %s\n",
0717                ev, sess->sessname);
0718     }
0719 }
0720 
0721 static void rnbd_init_cpu_qlists(struct rnbd_cpu_qlist __percpu *cpu_queues)
0722 {
0723     unsigned int cpu;
0724     struct rnbd_cpu_qlist *cpu_q;
0725 
0726     for_each_possible_cpu(cpu) {
0727         cpu_q = per_cpu_ptr(cpu_queues, cpu);
0728 
0729         cpu_q->cpu = cpu;
0730         INIT_LIST_HEAD(&cpu_q->requeue_list);
0731         spin_lock_init(&cpu_q->requeue_lock);
0732     }
0733 }
0734 
0735 static void destroy_mq_tags(struct rnbd_clt_session *sess)
0736 {
0737     if (sess->tag_set.tags)
0738         blk_mq_free_tag_set(&sess->tag_set);
0739 }
0740 
0741 static inline void wake_up_rtrs_waiters(struct rnbd_clt_session *sess)
0742 {
0743     sess->rtrs_ready = true;
0744     wake_up_all(&sess->rtrs_waitq);
0745 }
0746 
0747 static void close_rtrs(struct rnbd_clt_session *sess)
0748 {
0749     might_sleep();
0750 
0751     if (!IS_ERR_OR_NULL(sess->rtrs)) {
0752         rtrs_clt_close(sess->rtrs);
0753         sess->rtrs = NULL;
0754         wake_up_rtrs_waiters(sess);
0755     }
0756 }
0757 
0758 static void free_sess(struct rnbd_clt_session *sess)
0759 {
0760     WARN_ON(!list_empty(&sess->devs_list));
0761 
0762     might_sleep();
0763 
0764     close_rtrs(sess);
0765     destroy_mq_tags(sess);
0766     if (!list_empty(&sess->list)) {
0767         mutex_lock(&sess_lock);
0768         list_del(&sess->list);
0769         mutex_unlock(&sess_lock);
0770     }
0771     free_percpu(sess->cpu_queues);
0772     free_percpu(sess->cpu_rr);
0773     mutex_destroy(&sess->lock);
0774     kfree(sess);
0775 }
0776 
0777 static struct rnbd_clt_session *alloc_sess(const char *sessname)
0778 {
0779     struct rnbd_clt_session *sess;
0780     int err, cpu;
0781 
0782     sess = kzalloc_node(sizeof(*sess), GFP_KERNEL, NUMA_NO_NODE);
0783     if (!sess)
0784         return ERR_PTR(-ENOMEM);
0785     strscpy(sess->sessname, sessname, sizeof(sess->sessname));
0786     atomic_set(&sess->busy, 0);
0787     mutex_init(&sess->lock);
0788     INIT_LIST_HEAD(&sess->devs_list);
0789     INIT_LIST_HEAD(&sess->list);
0790     bitmap_zero(sess->cpu_queues_bm, num_possible_cpus());
0791     init_waitqueue_head(&sess->rtrs_waitq);
0792     refcount_set(&sess->refcount, 1);
0793 
0794     sess->cpu_queues = alloc_percpu(struct rnbd_cpu_qlist);
0795     if (!sess->cpu_queues) {
0796         err = -ENOMEM;
0797         goto err;
0798     }
0799     rnbd_init_cpu_qlists(sess->cpu_queues);
0800 
0801     /*
0802      * That is simple percpu variable which stores cpu indices, which are
0803      * incremented on each access.  We need that for the sake of fairness
0804      * to wake up queues in a round-robin manner.
0805      */
0806     sess->cpu_rr = alloc_percpu(int);
0807     if (!sess->cpu_rr) {
0808         err = -ENOMEM;
0809         goto err;
0810     }
0811     for_each_possible_cpu(cpu)
0812         * per_cpu_ptr(sess->cpu_rr, cpu) = cpu;
0813 
0814     return sess;
0815 
0816 err:
0817     free_sess(sess);
0818 
0819     return ERR_PTR(err);
0820 }
0821 
0822 static int wait_for_rtrs_connection(struct rnbd_clt_session *sess)
0823 {
0824     wait_event(sess->rtrs_waitq, sess->rtrs_ready);
0825     if (IS_ERR_OR_NULL(sess->rtrs))
0826         return -ECONNRESET;
0827 
0828     return 0;
0829 }
0830 
0831 static void wait_for_rtrs_disconnection(struct rnbd_clt_session *sess)
0832     __releases(&sess_lock)
0833     __acquires(&sess_lock)
0834 {
0835     DEFINE_WAIT(wait);
0836 
0837     prepare_to_wait(&sess->rtrs_waitq, &wait, TASK_UNINTERRUPTIBLE);
0838     if (IS_ERR_OR_NULL(sess->rtrs)) {
0839         finish_wait(&sess->rtrs_waitq, &wait);
0840         return;
0841     }
0842     mutex_unlock(&sess_lock);
0843     /* loop in caller, see __find_and_get_sess().
0844      * You can't leave mutex locked and call schedule(), you will catch a
0845      * deadlock with a caller of free_sess(), which has just put the last
0846      * reference and is about to take the sess_lock in order to delete
0847      * the session from the list.
0848      */
0849     schedule();
0850     mutex_lock(&sess_lock);
0851 }
0852 
0853 static struct rnbd_clt_session *__find_and_get_sess(const char *sessname)
0854     __releases(&sess_lock)
0855     __acquires(&sess_lock)
0856 {
0857     struct rnbd_clt_session *sess, *sn;
0858     int err;
0859 
0860 again:
0861     list_for_each_entry_safe(sess, sn, &sess_list, list) {
0862         if (strcmp(sessname, sess->sessname))
0863             continue;
0864 
0865         if (sess->rtrs_ready && IS_ERR_OR_NULL(sess->rtrs))
0866             /*
0867              * No RTRS connection, session is dying.
0868              */
0869             continue;
0870 
0871         if (rnbd_clt_get_sess(sess)) {
0872             /*
0873              * Alive session is found, wait for RTRS connection.
0874              */
0875             mutex_unlock(&sess_lock);
0876             err = wait_for_rtrs_connection(sess);
0877             if (err)
0878                 rnbd_clt_put_sess(sess);
0879             mutex_lock(&sess_lock);
0880 
0881             if (err)
0882                 /* Session is dying, repeat the loop */
0883                 goto again;
0884 
0885             return sess;
0886         }
0887         /*
0888          * Ref is 0, session is dying, wait for RTRS disconnect
0889          * in order to avoid session names clashes.
0890          */
0891         wait_for_rtrs_disconnection(sess);
0892         /*
0893          * RTRS is disconnected and soon session will be freed,
0894          * so repeat a loop.
0895          */
0896         goto again;
0897     }
0898 
0899     return NULL;
0900 }
0901 
0902 /* caller is responsible for initializing 'first' to false */
0903 static struct
0904 rnbd_clt_session *find_or_create_sess(const char *sessname, bool *first)
0905 {
0906     struct rnbd_clt_session *sess = NULL;
0907 
0908     mutex_lock(&sess_lock);
0909     sess = __find_and_get_sess(sessname);
0910     if (!sess) {
0911         sess = alloc_sess(sessname);
0912         if (IS_ERR(sess)) {
0913             mutex_unlock(&sess_lock);
0914             return sess;
0915         }
0916         list_add(&sess->list, &sess_list);
0917         *first = true;
0918     }
0919     mutex_unlock(&sess_lock);
0920 
0921     return sess;
0922 }
0923 
0924 static int rnbd_client_open(struct block_device *block_device, fmode_t mode)
0925 {
0926     struct rnbd_clt_dev *dev = block_device->bd_disk->private_data;
0927 
0928     if (get_disk_ro(dev->gd) && (mode & FMODE_WRITE))
0929         return -EPERM;
0930 
0931     if (dev->dev_state == DEV_STATE_UNMAPPED ||
0932         !rnbd_clt_get_dev(dev))
0933         return -EIO;
0934 
0935     return 0;
0936 }
0937 
0938 static void rnbd_client_release(struct gendisk *gen, fmode_t mode)
0939 {
0940     struct rnbd_clt_dev *dev = gen->private_data;
0941 
0942     rnbd_clt_put_dev(dev);
0943 }
0944 
0945 static int rnbd_client_getgeo(struct block_device *block_device,
0946                   struct hd_geometry *geo)
0947 {
0948     u64 size;
0949     struct rnbd_clt_dev *dev = block_device->bd_disk->private_data;
0950     struct queue_limits *limit = &dev->queue->limits;
0951 
0952     size = dev->size * (limit->logical_block_size / SECTOR_SIZE);
0953     geo->cylinders  = size >> 6;    /* size/64 */
0954     geo->heads  = 4;
0955     geo->sectors    = 16;
0956     geo->start  = 0;
0957 
0958     return 0;
0959 }
0960 
0961 static const struct block_device_operations rnbd_client_ops = {
0962     .owner      = THIS_MODULE,
0963     .open       = rnbd_client_open,
0964     .release    = rnbd_client_release,
0965     .getgeo     = rnbd_client_getgeo
0966 };
0967 
0968 /* The amount of data that belongs to an I/O and the amount of data that
0969  * should be read or written to the disk (bi_size) can differ.
0970  *
0971  * E.g. When WRITE_SAME is used, only a small amount of data is
0972  * transferred that is then written repeatedly over a lot of sectors.
0973  *
0974  * Get the size of data to be transferred via RTRS by summing up the size
0975  * of the scather-gather list entries.
0976  */
0977 static size_t rnbd_clt_get_sg_size(struct scatterlist *sglist, u32 len)
0978 {
0979     struct scatterlist *sg;
0980     size_t tsize = 0;
0981     int i;
0982 
0983     for_each_sg(sglist, sg, len, i)
0984         tsize += sg->length;
0985     return tsize;
0986 }
0987 
0988 static int rnbd_client_xfer_request(struct rnbd_clt_dev *dev,
0989                      struct request *rq,
0990                      struct rnbd_iu *iu)
0991 {
0992     struct rtrs_clt_sess *rtrs = dev->sess->rtrs;
0993     struct rtrs_permit *permit = iu->permit;
0994     struct rnbd_msg_io msg;
0995     struct rtrs_clt_req_ops req_ops;
0996     unsigned int sg_cnt = 0;
0997     struct kvec vec;
0998     size_t size;
0999     int err;
1000 
1001     iu->rq      = rq;
1002     iu->dev     = dev;
1003     msg.sector  = cpu_to_le64(blk_rq_pos(rq));
1004     msg.bi_size = cpu_to_le32(blk_rq_bytes(rq));
1005     msg.rw      = cpu_to_le32(rq_to_rnbd_flags(rq));
1006     msg.prio    = cpu_to_le16(req_get_ioprio(rq));
1007 
1008     /*
1009      * We only support discards with single segment for now.
1010      * See queue limits.
1011      */
1012     if (req_op(rq) != REQ_OP_DISCARD)
1013         sg_cnt = blk_rq_map_sg(dev->queue, rq, iu->sgt.sgl);
1014 
1015     if (sg_cnt == 0)
1016         sg_mark_end(&iu->sgt.sgl[0]);
1017 
1018     msg.hdr.type    = cpu_to_le16(RNBD_MSG_IO);
1019     msg.device_id   = cpu_to_le32(dev->device_id);
1020 
1021     vec = (struct kvec) {
1022         .iov_base = &msg,
1023         .iov_len  = sizeof(msg)
1024     };
1025     size = rnbd_clt_get_sg_size(iu->sgt.sgl, sg_cnt);
1026     req_ops = (struct rtrs_clt_req_ops) {
1027         .priv = iu,
1028         .conf_fn = msg_io_conf,
1029     };
1030     err = rtrs_clt_request(rq_data_dir(rq), &req_ops, rtrs, permit,
1031                    &vec, 1, size, iu->sgt.sgl, sg_cnt);
1032     if (err) {
1033         rnbd_clt_err_rl(dev, "RTRS failed to transfer IO, err: %d\n",
1034                  err);
1035         return err;
1036     }
1037 
1038     return 0;
1039 }
1040 
1041 /**
1042  * rnbd_clt_dev_add_to_requeue() - add device to requeue if session is busy
1043  * @dev:    Device to be checked
1044  * @q:      Queue to be added to the requeue list if required
1045  *
1046  * Description:
1047  *     If session is busy, that means someone will requeue us when resources
1048  *     are freed.  If session is not doing anything - device is not added to
1049  *     the list and @false is returned.
1050  */
1051 static bool rnbd_clt_dev_add_to_requeue(struct rnbd_clt_dev *dev,
1052                         struct rnbd_queue *q)
1053 {
1054     struct rnbd_clt_session *sess = dev->sess;
1055     struct rnbd_cpu_qlist *cpu_q;
1056     unsigned long flags;
1057     bool added = true;
1058     bool need_set;
1059 
1060     cpu_q = get_cpu_ptr(sess->cpu_queues);
1061     spin_lock_irqsave(&cpu_q->requeue_lock, flags);
1062 
1063     if (!test_and_set_bit_lock(0, &q->in_list)) {
1064         if (WARN_ON(!list_empty(&q->requeue_list)))
1065             goto unlock;
1066 
1067         need_set = !test_bit(cpu_q->cpu, sess->cpu_queues_bm);
1068         if (need_set) {
1069             set_bit(cpu_q->cpu, sess->cpu_queues_bm);
1070             /* Paired with rnbd_put_permit(). Set a bit first
1071              * and then observe the busy counter.
1072              */
1073             smp_mb__before_atomic();
1074         }
1075         if (atomic_read(&sess->busy)) {
1076             list_add_tail(&q->requeue_list, &cpu_q->requeue_list);
1077         } else {
1078             /* Very unlikely, but possible: busy counter was
1079              * observed as zero.  Drop all bits and return
1080              * false to restart the queue by ourselves.
1081              */
1082             if (need_set)
1083                 clear_bit(cpu_q->cpu, sess->cpu_queues_bm);
1084             clear_bit_unlock(0, &q->in_list);
1085             added = false;
1086         }
1087     }
1088 unlock:
1089     spin_unlock_irqrestore(&cpu_q->requeue_lock, flags);
1090     put_cpu_ptr(sess->cpu_queues);
1091 
1092     return added;
1093 }
1094 
1095 static void rnbd_clt_dev_kick_mq_queue(struct rnbd_clt_dev *dev,
1096                     struct blk_mq_hw_ctx *hctx,
1097                     int delay)
1098 {
1099     struct rnbd_queue *q = hctx->driver_data;
1100 
1101     if (delay != RNBD_DELAY_IFBUSY)
1102         blk_mq_delay_run_hw_queue(hctx, delay);
1103     else if (!rnbd_clt_dev_add_to_requeue(dev, q))
1104         /*
1105          * If session is not busy we have to restart
1106          * the queue ourselves.
1107          */
1108         blk_mq_delay_run_hw_queue(hctx, 10/*ms*/);
1109 }
1110 
1111 static blk_status_t rnbd_queue_rq(struct blk_mq_hw_ctx *hctx,
1112                    const struct blk_mq_queue_data *bd)
1113 {
1114     struct request *rq = bd->rq;
1115     struct rnbd_clt_dev *dev = rq->q->disk->private_data;
1116     struct rnbd_iu *iu = blk_mq_rq_to_pdu(rq);
1117     int err;
1118     blk_status_t ret = BLK_STS_IOERR;
1119 
1120     if (dev->dev_state != DEV_STATE_MAPPED)
1121         return BLK_STS_IOERR;
1122 
1123     iu->permit = rnbd_get_permit(dev->sess, RTRS_IO_CON,
1124                       RTRS_PERMIT_NOWAIT);
1125     if (!iu->permit) {
1126         rnbd_clt_dev_kick_mq_queue(dev, hctx, RNBD_DELAY_IFBUSY);
1127         return BLK_STS_RESOURCE;
1128     }
1129 
1130     iu->sgt.sgl = iu->first_sgl;
1131     err = sg_alloc_table_chained(&iu->sgt,
1132                      /* Even-if the request has no segment,
1133                       * sglist must have one entry at least.
1134                       */
1135                      blk_rq_nr_phys_segments(rq) ? : 1,
1136                      iu->sgt.sgl,
1137                      RNBD_INLINE_SG_CNT);
1138     if (err) {
1139         rnbd_clt_err_rl(dev, "sg_alloc_table_chained ret=%d\n", err);
1140         rnbd_clt_dev_kick_mq_queue(dev, hctx, 10/*ms*/);
1141         rnbd_put_permit(dev->sess, iu->permit);
1142         return BLK_STS_RESOURCE;
1143     }
1144 
1145     blk_mq_start_request(rq);
1146     err = rnbd_client_xfer_request(dev, rq, iu);
1147     if (err == 0)
1148         return BLK_STS_OK;
1149     if (err == -EAGAIN || err == -ENOMEM) {
1150         rnbd_clt_dev_kick_mq_queue(dev, hctx, 10/*ms*/);
1151         ret = BLK_STS_RESOURCE;
1152     }
1153     sg_free_table_chained(&iu->sgt, RNBD_INLINE_SG_CNT);
1154     rnbd_put_permit(dev->sess, iu->permit);
1155     return ret;
1156 }
1157 
1158 static int rnbd_rdma_poll(struct blk_mq_hw_ctx *hctx, struct io_comp_batch *iob)
1159 {
1160     struct rnbd_queue *q = hctx->driver_data;
1161     struct rnbd_clt_dev *dev = q->dev;
1162     int cnt;
1163 
1164     cnt = rtrs_clt_rdma_cq_direct(dev->sess->rtrs, hctx->queue_num);
1165     return cnt;
1166 }
1167 
1168 static int rnbd_rdma_map_queues(struct blk_mq_tag_set *set)
1169 {
1170     struct rnbd_clt_session *sess = set->driver_data;
1171 
1172     /* shared read/write queues */
1173     set->map[HCTX_TYPE_DEFAULT].nr_queues = num_online_cpus();
1174     set->map[HCTX_TYPE_DEFAULT].queue_offset = 0;
1175     set->map[HCTX_TYPE_READ].nr_queues = num_online_cpus();
1176     set->map[HCTX_TYPE_READ].queue_offset = 0;
1177     blk_mq_map_queues(&set->map[HCTX_TYPE_DEFAULT]);
1178     blk_mq_map_queues(&set->map[HCTX_TYPE_READ]);
1179 
1180     if (sess->nr_poll_queues) {
1181         /* dedicated queue for poll */
1182         set->map[HCTX_TYPE_POLL].nr_queues = sess->nr_poll_queues;
1183         set->map[HCTX_TYPE_POLL].queue_offset = set->map[HCTX_TYPE_READ].queue_offset +
1184             set->map[HCTX_TYPE_READ].nr_queues;
1185         blk_mq_map_queues(&set->map[HCTX_TYPE_POLL]);
1186         pr_info("[session=%s] mapped %d/%d/%d default/read/poll queues.\n",
1187             sess->sessname,
1188             set->map[HCTX_TYPE_DEFAULT].nr_queues,
1189             set->map[HCTX_TYPE_READ].nr_queues,
1190             set->map[HCTX_TYPE_POLL].nr_queues);
1191     } else {
1192         pr_info("[session=%s] mapped %d/%d default/read queues.\n",
1193             sess->sessname,
1194             set->map[HCTX_TYPE_DEFAULT].nr_queues,
1195             set->map[HCTX_TYPE_READ].nr_queues);
1196     }
1197 
1198     return 0;
1199 }
1200 
1201 static struct blk_mq_ops rnbd_mq_ops = {
1202     .queue_rq   = rnbd_queue_rq,
1203     .complete   = rnbd_softirq_done_fn,
1204     .map_queues     = rnbd_rdma_map_queues,
1205     .poll           = rnbd_rdma_poll,
1206 };
1207 
1208 static int setup_mq_tags(struct rnbd_clt_session *sess)
1209 {
1210     struct blk_mq_tag_set *tag_set = &sess->tag_set;
1211 
1212     memset(tag_set, 0, sizeof(*tag_set));
1213     tag_set->ops        = &rnbd_mq_ops;
1214     tag_set->queue_depth    = sess->queue_depth;
1215     tag_set->numa_node      = NUMA_NO_NODE;
1216     tag_set->flags      = BLK_MQ_F_SHOULD_MERGE |
1217                   BLK_MQ_F_TAG_QUEUE_SHARED;
1218     tag_set->cmd_size   = sizeof(struct rnbd_iu) + RNBD_RDMA_SGL_SIZE;
1219 
1220     /* for HCTX_TYPE_DEFAULT, HCTX_TYPE_READ, HCTX_TYPE_POLL */
1221     tag_set->nr_maps        = sess->nr_poll_queues ? HCTX_MAX_TYPES : 2;
1222     /*
1223      * HCTX_TYPE_DEFAULT and HCTX_TYPE_READ share one set of queues
1224      * others are for HCTX_TYPE_POLL
1225      */
1226     tag_set->nr_hw_queues   = num_online_cpus() + sess->nr_poll_queues;
1227     tag_set->driver_data    = sess;
1228 
1229     return blk_mq_alloc_tag_set(tag_set);
1230 }
1231 
1232 static struct rnbd_clt_session *
1233 find_and_get_or_create_sess(const char *sessname,
1234                 const struct rtrs_addr *paths,
1235                 size_t path_cnt, u16 port_nr, u32 nr_poll_queues)
1236 {
1237     struct rnbd_clt_session *sess;
1238     struct rtrs_attrs attrs;
1239     int err;
1240     bool first = false;
1241     struct rtrs_clt_ops rtrs_ops;
1242 
1243     sess = find_or_create_sess(sessname, &first);
1244     if (sess == ERR_PTR(-ENOMEM)) {
1245         return ERR_PTR(-ENOMEM);
1246     } else if ((nr_poll_queues && !first) ||  (!nr_poll_queues && sess->nr_poll_queues)) {
1247         /*
1248          * A device MUST have its own session to use the polling-mode.
1249          * It must fail to map new device with the same session.
1250          */
1251         err = -EINVAL;
1252         goto put_sess;
1253     }
1254 
1255     if (!first)
1256         return sess;
1257 
1258     if (!path_cnt) {
1259         pr_err("Session %s not found, and path parameter not given", sessname);
1260         err = -ENXIO;
1261         goto put_sess;
1262     }
1263 
1264     rtrs_ops = (struct rtrs_clt_ops) {
1265         .priv = sess,
1266         .link_ev = rnbd_clt_link_ev,
1267     };
1268     /*
1269      * Nothing was found, establish rtrs connection and proceed further.
1270      */
1271     sess->rtrs = rtrs_clt_open(&rtrs_ops, sessname,
1272                    paths, path_cnt, port_nr,
1273                    0, /* Do not use pdu of rtrs */
1274                    RECONNECT_DELAY,
1275                    MAX_RECONNECTS, nr_poll_queues);
1276     if (IS_ERR(sess->rtrs)) {
1277         err = PTR_ERR(sess->rtrs);
1278         goto wake_up_and_put;
1279     }
1280 
1281     err = rtrs_clt_query(sess->rtrs, &attrs);
1282     if (err)
1283         goto close_rtrs;
1284 
1285     sess->max_io_size = attrs.max_io_size;
1286     sess->queue_depth = attrs.queue_depth;
1287     sess->nr_poll_queues = nr_poll_queues;
1288     sess->max_segments = attrs.max_segments;
1289 
1290     err = setup_mq_tags(sess);
1291     if (err)
1292         goto close_rtrs;
1293 
1294     err = send_msg_sess_info(sess, RTRS_PERMIT_WAIT);
1295     if (err)
1296         goto close_rtrs;
1297 
1298     wake_up_rtrs_waiters(sess);
1299 
1300     return sess;
1301 
1302 close_rtrs:
1303     close_rtrs(sess);
1304 put_sess:
1305     rnbd_clt_put_sess(sess);
1306 
1307     return ERR_PTR(err);
1308 
1309 wake_up_and_put:
1310     wake_up_rtrs_waiters(sess);
1311     goto put_sess;
1312 }
1313 
1314 static inline void rnbd_init_hw_queue(struct rnbd_clt_dev *dev,
1315                        struct rnbd_queue *q,
1316                        struct blk_mq_hw_ctx *hctx)
1317 {
1318     INIT_LIST_HEAD(&q->requeue_list);
1319     q->dev  = dev;
1320     q->hctx = hctx;
1321 }
1322 
1323 static void rnbd_init_mq_hw_queues(struct rnbd_clt_dev *dev)
1324 {
1325     unsigned long i;
1326     struct blk_mq_hw_ctx *hctx;
1327     struct rnbd_queue *q;
1328 
1329     queue_for_each_hw_ctx(dev->queue, hctx, i) {
1330         q = &dev->hw_queues[i];
1331         rnbd_init_hw_queue(dev, q, hctx);
1332         hctx->driver_data = q;
1333     }
1334 }
1335 
1336 static void setup_request_queue(struct rnbd_clt_dev *dev,
1337                 struct rnbd_msg_open_rsp *rsp)
1338 {
1339     blk_queue_logical_block_size(dev->queue,
1340                      le16_to_cpu(rsp->logical_block_size));
1341     blk_queue_physical_block_size(dev->queue,
1342                       le16_to_cpu(rsp->physical_block_size));
1343     blk_queue_max_hw_sectors(dev->queue,
1344                  dev->sess->max_io_size / SECTOR_SIZE);
1345 
1346     /*
1347      * we don't support discards to "discontiguous" segments
1348      * in on request
1349      */
1350     blk_queue_max_discard_segments(dev->queue, 1);
1351 
1352     blk_queue_max_discard_sectors(dev->queue,
1353                       le32_to_cpu(rsp->max_discard_sectors));
1354     dev->queue->limits.discard_granularity =
1355                     le32_to_cpu(rsp->discard_granularity);
1356     dev->queue->limits.discard_alignment =
1357                     le32_to_cpu(rsp->discard_alignment);
1358     if (le16_to_cpu(rsp->secure_discard))
1359         blk_queue_max_secure_erase_sectors(dev->queue,
1360                     le32_to_cpu(rsp->max_discard_sectors));
1361     blk_queue_flag_set(QUEUE_FLAG_SAME_COMP, dev->queue);
1362     blk_queue_flag_set(QUEUE_FLAG_SAME_FORCE, dev->queue);
1363     blk_queue_max_segments(dev->queue, dev->sess->max_segments);
1364     blk_queue_io_opt(dev->queue, dev->sess->max_io_size);
1365     blk_queue_virt_boundary(dev->queue, SZ_4K - 1);
1366     blk_queue_write_cache(dev->queue,
1367                   !!(rsp->cache_policy & RNBD_WRITEBACK),
1368                   !!(rsp->cache_policy & RNBD_FUA));
1369 }
1370 
1371 static int rnbd_clt_setup_gen_disk(struct rnbd_clt_dev *dev,
1372                    struct rnbd_msg_open_rsp *rsp, int idx)
1373 {
1374     int err;
1375 
1376     dev->gd->major      = rnbd_client_major;
1377     dev->gd->first_minor    = idx << RNBD_PART_BITS;
1378     dev->gd->minors     = 1 << RNBD_PART_BITS;
1379     dev->gd->fops       = &rnbd_client_ops;
1380     dev->gd->queue      = dev->queue;
1381     dev->gd->private_data   = dev;
1382     snprintf(dev->gd->disk_name, sizeof(dev->gd->disk_name), "rnbd%d",
1383          idx);
1384     pr_debug("disk_name=%s, capacity=%llu\n",
1385          dev->gd->disk_name,
1386          le64_to_cpu(rsp->nsectors) *
1387          (le16_to_cpu(rsp->logical_block_size) / SECTOR_SIZE));
1388 
1389     set_capacity(dev->gd, le64_to_cpu(rsp->nsectors));
1390 
1391     if (dev->access_mode == RNBD_ACCESS_RO)
1392         set_disk_ro(dev->gd, true);
1393 
1394     /*
1395      * Network device does not need rotational
1396      */
1397     blk_queue_flag_set(QUEUE_FLAG_NONROT, dev->queue);
1398     err = add_disk(dev->gd);
1399     if (err)
1400         put_disk(dev->gd);
1401 
1402     return err;
1403 }
1404 
1405 static int rnbd_client_setup_device(struct rnbd_clt_dev *dev,
1406                     struct rnbd_msg_open_rsp *rsp)
1407 {
1408     int idx = dev->clt_device_id;
1409 
1410     dev->size = le64_to_cpu(rsp->nsectors) *
1411             le16_to_cpu(rsp->logical_block_size);
1412 
1413     dev->gd = blk_mq_alloc_disk(&dev->sess->tag_set, dev);
1414     if (IS_ERR(dev->gd))
1415         return PTR_ERR(dev->gd);
1416     dev->queue = dev->gd->queue;
1417     rnbd_init_mq_hw_queues(dev);
1418 
1419     setup_request_queue(dev, rsp);
1420     return rnbd_clt_setup_gen_disk(dev, rsp, idx);
1421 }
1422 
1423 static struct rnbd_clt_dev *init_dev(struct rnbd_clt_session *sess,
1424                       enum rnbd_access_mode access_mode,
1425                       const char *pathname,
1426                       u32 nr_poll_queues)
1427 {
1428     struct rnbd_clt_dev *dev;
1429     int ret;
1430 
1431     dev = kzalloc_node(sizeof(*dev), GFP_KERNEL, NUMA_NO_NODE);
1432     if (!dev)
1433         return ERR_PTR(-ENOMEM);
1434 
1435     /*
1436      * nr_cpu_ids: the number of softirq queues
1437      * nr_poll_queues: the number of polling queues
1438      */
1439     dev->hw_queues = kcalloc(nr_cpu_ids + nr_poll_queues,
1440                  sizeof(*dev->hw_queues),
1441                  GFP_KERNEL);
1442     if (!dev->hw_queues) {
1443         ret = -ENOMEM;
1444         goto out_alloc;
1445     }
1446 
1447     ret = ida_alloc_max(&index_ida, 1 << (MINORBITS - RNBD_PART_BITS),
1448                 GFP_KERNEL);
1449     if (ret < 0) {
1450         pr_err("Failed to initialize device '%s' from session %s, allocating idr failed, err: %d\n",
1451                pathname, sess->sessname, ret);
1452         goto out_queues;
1453     }
1454 
1455     dev->pathname = kstrdup(pathname, GFP_KERNEL);
1456     if (!dev->pathname) {
1457         ret = -ENOMEM;
1458         goto out_queues;
1459     }
1460 
1461     dev->clt_device_id  = ret;
1462     dev->sess       = sess;
1463     dev->access_mode    = access_mode;
1464     dev->nr_poll_queues = nr_poll_queues;
1465     mutex_init(&dev->lock);
1466     refcount_set(&dev->refcount, 1);
1467     dev->dev_state = DEV_STATE_INIT;
1468 
1469     /*
1470      * Here we called from sysfs entry, thus clt-sysfs is
1471      * responsible that session will not disappear.
1472      */
1473     WARN_ON(!rnbd_clt_get_sess(sess));
1474 
1475     return dev;
1476 
1477 out_queues:
1478     kfree(dev->hw_queues);
1479 out_alloc:
1480     kfree(dev);
1481     return ERR_PTR(ret);
1482 }
1483 
1484 static bool __exists_dev(const char *pathname, const char *sessname)
1485 {
1486     struct rnbd_clt_session *sess;
1487     struct rnbd_clt_dev *dev;
1488     bool found = false;
1489 
1490     list_for_each_entry(sess, &sess_list, list) {
1491         if (sessname && strncmp(sess->sessname, sessname,
1492                     sizeof(sess->sessname)))
1493             continue;
1494         mutex_lock(&sess->lock);
1495         list_for_each_entry(dev, &sess->devs_list, list) {
1496             if (strlen(dev->pathname) == strlen(pathname) &&
1497                 !strcmp(dev->pathname, pathname)) {
1498                 found = true;
1499                 break;
1500             }
1501         }
1502         mutex_unlock(&sess->lock);
1503         if (found)
1504             break;
1505     }
1506 
1507     return found;
1508 }
1509 
1510 static bool exists_devpath(const char *pathname, const char *sessname)
1511 {
1512     bool found;
1513 
1514     mutex_lock(&sess_lock);
1515     found = __exists_dev(pathname, sessname);
1516     mutex_unlock(&sess_lock);
1517 
1518     return found;
1519 }
1520 
1521 static bool insert_dev_if_not_exists_devpath(struct rnbd_clt_dev *dev)
1522 {
1523     bool found;
1524     struct rnbd_clt_session *sess = dev->sess;
1525 
1526     mutex_lock(&sess_lock);
1527     found = __exists_dev(dev->pathname, sess->sessname);
1528     if (!found) {
1529         mutex_lock(&sess->lock);
1530         list_add_tail(&dev->list, &sess->devs_list);
1531         mutex_unlock(&sess->lock);
1532     }
1533     mutex_unlock(&sess_lock);
1534 
1535     return found;
1536 }
1537 
1538 static void delete_dev(struct rnbd_clt_dev *dev)
1539 {
1540     struct rnbd_clt_session *sess = dev->sess;
1541 
1542     mutex_lock(&sess->lock);
1543     list_del(&dev->list);
1544     mutex_unlock(&sess->lock);
1545 }
1546 
1547 struct rnbd_clt_dev *rnbd_clt_map_device(const char *sessname,
1548                        struct rtrs_addr *paths,
1549                        size_t path_cnt, u16 port_nr,
1550                        const char *pathname,
1551                        enum rnbd_access_mode access_mode,
1552                        u32 nr_poll_queues)
1553 {
1554     struct rnbd_clt_session *sess;
1555     struct rnbd_clt_dev *dev;
1556     int ret, errno;
1557     struct rnbd_msg_open_rsp *rsp;
1558     struct rnbd_msg_open msg;
1559     struct rnbd_iu *iu;
1560     struct kvec vec = {
1561         .iov_base = &msg,
1562         .iov_len  = sizeof(msg)
1563     };
1564 
1565     if (exists_devpath(pathname, sessname))
1566         return ERR_PTR(-EEXIST);
1567 
1568     sess = find_and_get_or_create_sess(sessname, paths, path_cnt, port_nr, nr_poll_queues);
1569     if (IS_ERR(sess))
1570         return ERR_CAST(sess);
1571 
1572     dev = init_dev(sess, access_mode, pathname, nr_poll_queues);
1573     if (IS_ERR(dev)) {
1574         pr_err("map_device: failed to map device '%s' from session %s, can't initialize device, err: %ld\n",
1575                pathname, sess->sessname, PTR_ERR(dev));
1576         ret = PTR_ERR(dev);
1577         goto put_sess;
1578     }
1579     if (insert_dev_if_not_exists_devpath(dev)) {
1580         ret = -EEXIST;
1581         goto put_dev;
1582     }
1583 
1584     rsp = kzalloc(sizeof(*rsp), GFP_KERNEL);
1585     if (!rsp) {
1586         ret = -ENOMEM;
1587         goto del_dev;
1588     }
1589 
1590     iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
1591     if (!iu) {
1592         ret = -ENOMEM;
1593         kfree(rsp);
1594         goto del_dev;
1595     }
1596     iu->buf = rsp;
1597     iu->dev = dev;
1598     sg_init_one(iu->sgt.sgl, rsp, sizeof(*rsp));
1599 
1600     msg.hdr.type    = cpu_to_le16(RNBD_MSG_OPEN);
1601     msg.access_mode = dev->access_mode;
1602     strscpy(msg.dev_name, dev->pathname, sizeof(msg.dev_name));
1603 
1604     WARN_ON(!rnbd_clt_get_dev(dev));
1605     ret = send_usr_msg(sess->rtrs, READ, iu,
1606                &vec, sizeof(*rsp), iu->sgt.sgl, 1,
1607                msg_open_conf, &errno, RTRS_PERMIT_WAIT);
1608     if (ret) {
1609         rnbd_clt_put_dev(dev);
1610         rnbd_put_iu(sess, iu);
1611     } else {
1612         ret = errno;
1613     }
1614     if (ret) {
1615         rnbd_clt_err(dev,
1616                   "map_device: failed, can't open remote device, err: %d\n",
1617                   ret);
1618         goto put_iu;
1619     }
1620     mutex_lock(&dev->lock);
1621     pr_debug("Opened remote device: session=%s, path='%s'\n",
1622          sess->sessname, pathname);
1623     ret = rnbd_client_setup_device(dev, rsp);
1624     if (ret) {
1625         rnbd_clt_err(dev,
1626                   "map_device: Failed to configure device, err: %d\n",
1627                   ret);
1628         mutex_unlock(&dev->lock);
1629         goto send_close;
1630     }
1631 
1632     rnbd_clt_info(dev,
1633                "map_device: Device mapped as %s (nsectors: %llu, logical_block_size: %d, physical_block_size: %d, max_discard_sectors: %d, discard_granularity: %d, discard_alignment: %d, secure_discard: %d, max_segments: %d, max_hw_sectors: %d, wc: %d, fua: %d)\n",
1634                dev->gd->disk_name, le64_to_cpu(rsp->nsectors),
1635                le16_to_cpu(rsp->logical_block_size),
1636                le16_to_cpu(rsp->physical_block_size),
1637                le32_to_cpu(rsp->max_discard_sectors),
1638                le32_to_cpu(rsp->discard_granularity),
1639                le32_to_cpu(rsp->discard_alignment),
1640                le16_to_cpu(rsp->secure_discard),
1641                sess->max_segments, sess->max_io_size / SECTOR_SIZE,
1642                !!(rsp->cache_policy & RNBD_WRITEBACK),
1643                !!(rsp->cache_policy & RNBD_FUA));
1644 
1645     mutex_unlock(&dev->lock);
1646     kfree(rsp);
1647     rnbd_put_iu(sess, iu);
1648     rnbd_clt_put_sess(sess);
1649 
1650     return dev;
1651 
1652 send_close:
1653     send_msg_close(dev, dev->device_id, RTRS_PERMIT_WAIT);
1654 put_iu:
1655     kfree(rsp);
1656     rnbd_put_iu(sess, iu);
1657 del_dev:
1658     delete_dev(dev);
1659 put_dev:
1660     rnbd_clt_put_dev(dev);
1661 put_sess:
1662     rnbd_clt_put_sess(sess);
1663 
1664     return ERR_PTR(ret);
1665 }
1666 
1667 static void destroy_gen_disk(struct rnbd_clt_dev *dev)
1668 {
1669     del_gendisk(dev->gd);
1670     put_disk(dev->gd);
1671 }
1672 
1673 static void destroy_sysfs(struct rnbd_clt_dev *dev,
1674               const struct attribute *sysfs_self)
1675 {
1676     rnbd_clt_remove_dev_symlink(dev);
1677     if (dev->kobj.state_initialized) {
1678         if (sysfs_self)
1679             /* To avoid deadlock firstly remove itself */
1680             sysfs_remove_file_self(&dev->kobj, sysfs_self);
1681         kobject_del(&dev->kobj);
1682         kobject_put(&dev->kobj);
1683     }
1684 }
1685 
1686 int rnbd_clt_unmap_device(struct rnbd_clt_dev *dev, bool force,
1687                const struct attribute *sysfs_self)
1688 {
1689     struct rnbd_clt_session *sess = dev->sess;
1690     int refcount, ret = 0;
1691     bool was_mapped;
1692 
1693     mutex_lock(&dev->lock);
1694     if (dev->dev_state == DEV_STATE_UNMAPPED) {
1695         rnbd_clt_info(dev, "Device is already being unmapped\n");
1696         ret = -EALREADY;
1697         goto err;
1698     }
1699     refcount = refcount_read(&dev->refcount);
1700     if (!force && refcount > 1) {
1701         rnbd_clt_err(dev,
1702                   "Closing device failed, device is in use, (%d device users)\n",
1703                   refcount - 1);
1704         ret = -EBUSY;
1705         goto err;
1706     }
1707     was_mapped = (dev->dev_state == DEV_STATE_MAPPED);
1708     dev->dev_state = DEV_STATE_UNMAPPED;
1709     mutex_unlock(&dev->lock);
1710 
1711     delete_dev(dev);
1712     destroy_sysfs(dev, sysfs_self);
1713     destroy_gen_disk(dev);
1714     if (was_mapped && sess->rtrs)
1715         send_msg_close(dev, dev->device_id, RTRS_PERMIT_WAIT);
1716 
1717     rnbd_clt_info(dev, "Device is unmapped\n");
1718 
1719     /* Likely last reference put */
1720     rnbd_clt_put_dev(dev);
1721 
1722     /*
1723      * Here device and session can be vanished!
1724      */
1725 
1726     return 0;
1727 err:
1728     mutex_unlock(&dev->lock);
1729 
1730     return ret;
1731 }
1732 
1733 int rnbd_clt_remap_device(struct rnbd_clt_dev *dev)
1734 {
1735     int err;
1736 
1737     mutex_lock(&dev->lock);
1738     if (dev->dev_state == DEV_STATE_MAPPED_DISCONNECTED)
1739         err = 0;
1740     else if (dev->dev_state == DEV_STATE_UNMAPPED)
1741         err = -ENODEV;
1742     else if (dev->dev_state == DEV_STATE_MAPPED)
1743         err = -EALREADY;
1744     else
1745         err = -EBUSY;
1746     mutex_unlock(&dev->lock);
1747     if (!err) {
1748         rnbd_clt_info(dev, "Remapping device.\n");
1749         err = send_msg_open(dev, RTRS_PERMIT_WAIT);
1750         if (err)
1751             rnbd_clt_err(dev, "remap_device: %d\n", err);
1752     }
1753 
1754     return err;
1755 }
1756 
1757 static void unmap_device_work(struct work_struct *work)
1758 {
1759     struct rnbd_clt_dev *dev;
1760 
1761     dev = container_of(work, typeof(*dev), unmap_on_rmmod_work);
1762     rnbd_clt_unmap_device(dev, true, NULL);
1763 }
1764 
1765 static void rnbd_destroy_sessions(void)
1766 {
1767     struct rnbd_clt_session *sess, *sn;
1768     struct rnbd_clt_dev *dev, *tn;
1769 
1770     /* Firstly forbid access through sysfs interface */
1771     rnbd_clt_destroy_sysfs_files();
1772 
1773     /*
1774      * Here at this point there is no any concurrent access to sessions
1775      * list and devices list:
1776      *   1. New session or device can't be created - session sysfs files
1777      *      are removed.
1778      *   2. Device or session can't be removed - module reference is taken
1779      *      into account in unmap device sysfs callback.
1780      *   3. No IO requests inflight - each file open of block_dev increases
1781      *      module reference in get_disk().
1782      *
1783      * But still there can be user requests inflights, which are sent by
1784      * asynchronous send_msg_*() functions, thus before unmapping devices
1785      * RTRS session must be explicitly closed.
1786      */
1787 
1788     list_for_each_entry_safe(sess, sn, &sess_list, list) {
1789         if (!rnbd_clt_get_sess(sess))
1790             continue;
1791         close_rtrs(sess);
1792         list_for_each_entry_safe(dev, tn, &sess->devs_list, list) {
1793             /*
1794              * Here unmap happens in parallel for only one reason:
1795              * del_gendisk() takes around half a second, so
1796              * on huge amount of devices the whole module unload
1797              * procedure takes minutes.
1798              */
1799             INIT_WORK(&dev->unmap_on_rmmod_work, unmap_device_work);
1800             queue_work(rnbd_clt_wq, &dev->unmap_on_rmmod_work);
1801         }
1802         rnbd_clt_put_sess(sess);
1803     }
1804     /* Wait for all scheduled unmap works */
1805     flush_workqueue(rnbd_clt_wq);
1806     WARN_ON(!list_empty(&sess_list));
1807 }
1808 
1809 static int __init rnbd_client_init(void)
1810 {
1811     int err = 0;
1812 
1813     BUILD_BUG_ON(sizeof(struct rnbd_msg_hdr) != 4);
1814     BUILD_BUG_ON(sizeof(struct rnbd_msg_sess_info) != 36);
1815     BUILD_BUG_ON(sizeof(struct rnbd_msg_sess_info_rsp) != 36);
1816     BUILD_BUG_ON(sizeof(struct rnbd_msg_open) != 264);
1817     BUILD_BUG_ON(sizeof(struct rnbd_msg_close) != 8);
1818     BUILD_BUG_ON(sizeof(struct rnbd_msg_open_rsp) != 56);
1819     rnbd_client_major = register_blkdev(rnbd_client_major, "rnbd");
1820     if (rnbd_client_major <= 0) {
1821         pr_err("Failed to load module, block device registration failed\n");
1822         return -EBUSY;
1823     }
1824 
1825     err = rnbd_clt_create_sysfs_files();
1826     if (err) {
1827         pr_err("Failed to load module, creating sysfs device files failed, err: %d\n",
1828                err);
1829         unregister_blkdev(rnbd_client_major, "rnbd");
1830         return err;
1831     }
1832     rnbd_clt_wq = alloc_workqueue("rnbd_clt_wq", 0, 0);
1833     if (!rnbd_clt_wq) {
1834         pr_err("Failed to load module, alloc_workqueue failed.\n");
1835         rnbd_clt_destroy_sysfs_files();
1836         unregister_blkdev(rnbd_client_major, "rnbd");
1837         err = -ENOMEM;
1838     }
1839 
1840     return err;
1841 }
1842 
1843 static void __exit rnbd_client_exit(void)
1844 {
1845     rnbd_destroy_sessions();
1846     unregister_blkdev(rnbd_client_major, "rnbd");
1847     ida_destroy(&index_ida);
1848     destroy_workqueue(rnbd_clt_wq);
1849 }
1850 
1851 module_init(rnbd_client_init);
1852 module_exit(rnbd_client_exit);
[ Source navigation ] [ Diff markup ] [ Identifier search ] [ general search ]

This page was automatically generated by the 2.1.0 LXR engine.

The LXR team
Valid CSS 2.1! Valid HTML 4.01!