drivers/block/rnbd/rnbd-clt.c

   1 // SPDX-License-Identifier: GPL-2.0-or-later
   2 /*
   3  * RDMA Network Block Driver
   4  *
   5  * Copyright (c) 2014 - 2018 ProfitBricks GmbH. All rights reserved.
   6  * Copyright (c) 2018 - 2019 1&1 IONOS Cloud GmbH. All rights reserved.
   7  * Copyright (c) 2019 - 2020 1&1 IONOS SE. All rights reserved.
   8  */
   9
  10 #undef pr_fmt
  11 #define pr_fmt(fmt) KBUILD_MODNAME " L" __stringify(__LINE__) ": " fmt
  12
  13 #include <linux/module.h>
  14 #include <linux/blkdev.h>
  15 #include <linux/hdreg.h>
  16 #include <linux/scatterlist.h>
  17 #include <linux/idr.h>
  18
  19 #include "rnbd-clt.h"
  20
  21 MODULE_DESCRIPTION("RDMA Network Block Device Client");
  22 MODULE_LICENSE("GPL");
  23
  24 static int rnbd_client_major;
  25 static DEFINE_IDA(index_ida);
  26 static DEFINE_MUTEX(ida_lock);
  27 static DEFINE_MUTEX(sess_lock);
  28 static LIST_HEAD(sess_list);
  29
  30 /*
  31  * Maximum number of partitions an instance can have.
  32  * 6 bits = 64 minors = 63 partitions (one minor is used for the device itself)
  33  */
  34 #define RNBD_PART_BITS          6
  35
  36 static inline bool rnbd_clt_get_sess(struct rnbd_clt_session *sess)
  37 {
  38         return refcount_inc_not_zero(&sess->refcount);
  39 }
  40
  41 static void free_sess(struct rnbd_clt_session *sess);
  42
  43 static void rnbd_clt_put_sess(struct rnbd_clt_session *sess)
  44 {
  45         might_sleep();
  46
  47         if (refcount_dec_and_test(&sess->refcount))
  48                 free_sess(sess);
  49 }
  50
  51 static void rnbd_clt_put_dev(struct rnbd_clt_dev *dev)
  52 {
  53         might_sleep();
  54
  55         if (!refcount_dec_and_test(&dev->refcount))
  56                 return;
  57
  58         mutex_lock(&ida_lock);
  59         ida_simple_remove(&index_ida, dev->clt_device_id);
  60         mutex_unlock(&ida_lock);
  61         kfree(dev->hw_queues);
  62         kfree(dev->pathname);
  63         rnbd_clt_put_sess(dev->sess);
  64         mutex_destroy(&dev->lock);
  65         kfree(dev);
  66 }
  67
  68 static inline bool rnbd_clt_get_dev(struct rnbd_clt_dev *dev)
  69 {
  70         return refcount_inc_not_zero(&dev->refcount);
  71 }
  72
  73 static int rnbd_clt_set_dev_attr(struct rnbd_clt_dev *dev,
  74                                  const struct rnbd_msg_open_rsp *rsp)
  75 {
  76         struct rnbd_clt_session *sess = dev->sess;
  77
  78         if (!rsp->logical_block_size)
  79                 return -EINVAL;
  80
  81         dev->device_id              = le32_to_cpu(rsp->device_id);
  82         dev->nsectors               = le64_to_cpu(rsp->nsectors);
  83         dev->logical_block_size     = le16_to_cpu(rsp->logical_block_size);
  84         dev->physical_block_size    = le16_to_cpu(rsp->physical_block_size);
  85         dev->max_write_same_sectors = le32_to_cpu(rsp->max_write_same_sectors);
  86         dev->max_discard_sectors    = le32_to_cpu(rsp->max_discard_sectors);
  87         dev->discard_granularity    = le32_to_cpu(rsp->discard_granularity);
  88         dev->discard_alignment      = le32_to_cpu(rsp->discard_alignment);
  89         dev->secure_discard         = le16_to_cpu(rsp->secure_discard);
  90         dev->rotational             = rsp->rotational;
  91         dev->wc                     = !!(rsp->cache_policy & RNBD_WRITEBACK);
  92         dev->fua                    = !!(rsp->cache_policy & RNBD_FUA);
  93
  94         dev->max_hw_sectors = sess->max_io_size / SECTOR_SIZE;
  95         dev->max_segments = BMAX_SEGMENTS;
  96
  97         return 0;
  98 }
  99
 100 static int rnbd_clt_change_capacity(struct rnbd_clt_dev *dev,
 101                                     size_t new_nsectors)
 102 {
 103         rnbd_clt_info(dev, "Device size changed from %zu to %zu sectors\n",
 104                        dev->nsectors, new_nsectors);
 105         dev->nsectors = new_nsectors;
 106         set_capacity_and_notify(dev->gd, dev->nsectors);
 107         return 0;
 108 }
 109
 110 static int process_msg_open_rsp(struct rnbd_clt_dev *dev,
 111                                 struct rnbd_msg_open_rsp *rsp)
 112 {
 113         struct kobject *gd_kobj;
 114         int err = 0;
 115
 116         mutex_lock(&dev->lock);
 117         if (dev->dev_state == DEV_STATE_UNMAPPED) {
 118                 rnbd_clt_info(dev,
 119                                "Ignoring Open-Response message from server for  unmapped device\n");
 120                 err = -ENOENT;
 121                 goto out;
 122         }
 123         if (dev->dev_state == DEV_STATE_MAPPED_DISCONNECTED) {
 124                 u64 nsectors = le64_to_cpu(rsp->nsectors);
 125
 126                 /*
 127                  * If the device was remapped and the size changed in the
 128                  * meantime we need to revalidate it
 129                  */
 130                 if (dev->nsectors != nsectors)
 131                         rnbd_clt_change_capacity(dev, nsectors);
 132                 gd_kobj = &disk_to_dev(dev->gd)->kobj;
 133                 kobject_uevent(gd_kobj, KOBJ_ONLINE);
 134                 rnbd_clt_info(dev, "Device online, device remapped successfully\n");
 135         }
 136         err = rnbd_clt_set_dev_attr(dev, rsp);
 137         if (err)
 138                 goto out;
 139         dev->dev_state = DEV_STATE_MAPPED;
 140
 141 out:
 142         mutex_unlock(&dev->lock);
 143
 144         return err;
 145 }
 146
 147 int rnbd_clt_resize_disk(struct rnbd_clt_dev *dev, size_t newsize)
 148 {
 149         int ret = 0;
 150
 151         mutex_lock(&dev->lock);
 152         if (dev->dev_state != DEV_STATE_MAPPED) {
 153                 pr_err("Failed to set new size of the device, device is not opened\n");
 154                 ret = -ENOENT;
 155                 goto out;
 156         }
 157         ret = rnbd_clt_change_capacity(dev, newsize);
 158
 159 out:
 160         mutex_unlock(&dev->lock);
 161
 162         return ret;
 163 }
 164
 165 static inline void rnbd_clt_dev_requeue(struct rnbd_queue *q)
 166 {
 167         if (WARN_ON(!q->hctx))
 168                 return;
 169
 170         /* We can come here from interrupt, thus async=true */
 171         blk_mq_run_hw_queue(q->hctx, true);
 172 }
 173
 174 enum {
 175         RNBD_DELAY_IFBUSY = -1,
 176 };
 177
 178 /**
 179  * rnbd_get_cpu_qlist() - finds a list with HW queues to be rerun
 180  * @sess:       Session to find a queue for
 181  * @cpu:        Cpu to start the search from
 182  *
 183  * Description:
 184  *     Each CPU has a list of HW queues, which needs to be rerun.  If a list
 185  *     is not empty - it is marked with a bit.  This function finds first
 186  *     set bit in a bitmap and returns corresponding CPU list.
 187  */
 188 static struct rnbd_cpu_qlist *
 189 rnbd_get_cpu_qlist(struct rnbd_clt_session *sess, int cpu)
 190 {
 191         int bit;
 192
 193         /* Search from cpu to nr_cpu_ids */
 194         bit = find_next_bit(sess->cpu_queues_bm, nr_cpu_ids, cpu);
 195         if (bit < nr_cpu_ids) {
 196                 return per_cpu_ptr(sess->cpu_queues, bit);
 197         } else if (cpu != 0) {
 198                 /* Search from 0 to cpu */
 199                 bit = find_next_bit(sess->cpu_queues_bm, cpu, 0);
 200                 if (bit < cpu)
 201                         return per_cpu_ptr(sess->cpu_queues, bit);
 202         }
 203
 204         return NULL;
 205 }
 206
 207 static inline int nxt_cpu(int cpu)
 208 {
 209         return (cpu + 1) % nr_cpu_ids;
 210 }
 211
 212 /**
 213  * rnbd_rerun_if_needed() - rerun next queue marked as stopped
 214  * @sess:       Session to rerun a queue on
 215  *
 216  * Description:
 217  *     Each CPU has it's own list of HW queues, which should be rerun.
 218  *     Function finds such list with HW queues, takes a list lock, picks up
 219  *     the first HW queue out of the list and requeues it.
 220  *
 221  * Return:
 222  *     True if the queue was requeued, false otherwise.
 223  *
 224  * Context:
 225  *     Does not matter.
 226  */
 227 static bool rnbd_rerun_if_needed(struct rnbd_clt_session *sess)
 228 {
 229         struct rnbd_queue *q = NULL;
 230         struct rnbd_cpu_qlist *cpu_q;
 231         unsigned long flags;
 232         int *cpup;
 233
 234         /*
 235          * To keep fairness and not to let other queues starve we always
 236          * try to wake up someone else in round-robin manner.  That of course
 237          * increases latency but queues always have a chance to be executed.
 238          */
 239         cpup = get_cpu_ptr(sess->cpu_rr);
 240         for (cpu_q = rnbd_get_cpu_qlist(sess, nxt_cpu(*cpup)); cpu_q;
 241              cpu_q = rnbd_get_cpu_qlist(sess, nxt_cpu(cpu_q->cpu))) {
 242                 if (!spin_trylock_irqsave(&cpu_q->requeue_lock, flags))
 243                         continue;
 244                 if (unlikely(!test_bit(cpu_q->cpu, sess->cpu_queues_bm)))
 245                         goto unlock;
 246                 q = list_first_entry_or_null(&cpu_q->requeue_list,
 247                                              typeof(*q), requeue_list);
 248                 if (WARN_ON(!q))
 249                         goto clear_bit;
 250                 list_del_init(&q->requeue_list);
 251                 clear_bit_unlock(0, &q->in_list);
 252
 253                 if (list_empty(&cpu_q->requeue_list)) {
 254                         /* Clear bit if nothing is left */
 255 clear_bit:
 256                         clear_bit(cpu_q->cpu, sess->cpu_queues_bm);
 257                 }
 258 unlock:
 259                 spin_unlock_irqrestore(&cpu_q->requeue_lock, flags);
 260
 261                 if (q)
 262                         break;
 263         }
 264
 265         /**
 266          * Saves the CPU that is going to be requeued on the per-cpu var. Just
 267          * incrementing it doesn't work because rnbd_get_cpu_qlist() will
 268          * always return the first CPU with something on the queue list when the
 269          * value stored on the var is greater than the last CPU with something
 270          * on the list.
 271          */
 272         if (cpu_q)
 273                 *cpup = cpu_q->cpu;
 274         put_cpu_var(sess->cpu_rr);
 275
 276         if (q)
 277                 rnbd_clt_dev_requeue(q);
 278
 279         return q;
 280 }
 281
 282 /**
 283  * rnbd_rerun_all_if_idle() - rerun all queues left in the list if
 284  *                               session is idling (there are no requests
 285  *                               in-flight).
 286  * @sess:       Session to rerun the queues on
 287  *
 288  * Description:
 289  *     This function tries to rerun all stopped queues if there are no
 290  *     requests in-flight anymore.  This function tries to solve an obvious
 291  *     problem, when number of tags < than number of queues (hctx), which
 292  *     are stopped and put to sleep.  If last permit, which has been just put,
 293  *     does not wake up all left queues (hctxs), IO requests hang forever.
 294  *
 295  *     That can happen when all number of permits, say N, have been exhausted
 296  *     from one CPU, and we have many block devices per session, say M.
 297  *     Each block device has it's own queue (hctx) for each CPU, so eventually
 298  *     we can put that number of queues (hctxs) to sleep: M x nr_cpu_ids.
 299  *     If number of permits N < M x nr_cpu_ids finally we will get an IO hang.
 300  *
 301  *     To avoid this hang last caller of rnbd_put_permit() (last caller is the
 302  *     one who observes sess->busy == 0) must wake up all remaining queues.
 303  *
 304  * Context:
 305  *     Does not matter.
 306  */
 307 static void rnbd_rerun_all_if_idle(struct rnbd_clt_session *sess)
 308 {
 309         bool requeued;
 310
 311         do {
 312                 requeued = rnbd_rerun_if_needed(sess);
 313         } while (atomic_read(&sess->busy) == 0 && requeued);
 314 }
 315
 316 static struct rtrs_permit *rnbd_get_permit(struct rnbd_clt_session *sess,
 317                                              enum rtrs_clt_con_type con_type,
 318                                              enum wait_type wait)
 319 {
 320         struct rtrs_permit *permit;
 321
 322         permit = rtrs_clt_get_permit(sess->rtrs, con_type, wait);
 323         if (likely(permit))
 324                 /* We have a subtle rare case here, when all permits can be
 325                  * consumed before busy counter increased.  This is safe,
 326                  * because loser will get NULL as a permit, observe 0 busy
 327                  * counter and immediately restart the queue himself.
 328                  */
 329                 atomic_inc(&sess->busy);
 330
 331         return permit;
 332 }
 333
 334 static void rnbd_put_permit(struct rnbd_clt_session *sess,
 335                              struct rtrs_permit *permit)
 336 {
 337         rtrs_clt_put_permit(sess->rtrs, permit);
 338         atomic_dec(&sess->busy);
 339         /* Paired with rnbd_clt_dev_add_to_requeue().  Decrement first
 340          * and then check queue bits.
 341          */
 342         smp_mb__after_atomic();
 343         rnbd_rerun_all_if_idle(sess);
 344 }
 345
 346 static struct rnbd_iu *rnbd_get_iu(struct rnbd_clt_session *sess,
 347                                      enum rtrs_clt_con_type con_type,
 348                                      enum wait_type wait)
 349 {
 350         struct rnbd_iu *iu;
 351         struct rtrs_permit *permit;
 352
 353         iu = kzalloc(sizeof(*iu), GFP_KERNEL);
 354         if (!iu) {
 355                 return NULL;
 356         }
 357
 358         permit = rnbd_get_permit(sess, con_type, wait);
 359         if (unlikely(!permit)) {
 360                 kfree(iu);
 361                 return NULL;
 362         }
 363
 364         iu->permit = permit;
 365         /*
 366          * 1st reference is dropped after finishing sending a "user" message,
 367          * 2nd reference is dropped after confirmation with the response is
 368          * returned.
 369          * 1st and 2nd can happen in any order, so the rnbd_iu should be
 370          * released (rtrs_permit returned to rtrs) only after both
 371          * are finished.
 372          */
 373         atomic_set(&iu->refcount, 2);
 374         init_waitqueue_head(&iu->comp.wait);
 375         iu->comp.errno = INT_MAX;
 376
 377         if (sg_alloc_table(&iu->sgt, 1, GFP_KERNEL)) {
 378                 rnbd_put_permit(sess, permit);
 379                 kfree(iu);
 380                 return NULL;
 381         }
 382
 383         return iu;
 384 }
 385
 386 static void rnbd_put_iu(struct rnbd_clt_session *sess, struct rnbd_iu *iu)
 387 {
 388         if (atomic_dec_and_test(&iu->refcount)) {
 389                 sg_free_table(&iu->sgt);
 390                 rnbd_put_permit(sess, iu->permit);
 391                 kfree(iu);
 392         }
 393 }
 394
 395 static void rnbd_softirq_done_fn(struct request *rq)
 396 {
 397         struct rnbd_clt_dev *dev        = rq->rq_disk->private_data;
 398         struct rnbd_clt_session *sess   = dev->sess;
 399         struct rnbd_iu *iu;
 400
 401         iu = blk_mq_rq_to_pdu(rq);
 402         sg_free_table_chained(&iu->sgt, RNBD_INLINE_SG_CNT);
 403         rnbd_put_permit(sess, iu->permit);
 404         blk_mq_end_request(rq, errno_to_blk_status(iu->errno));
 405 }
 406
 407 static void msg_io_conf(void *priv, int errno)
 408 {
 409         struct rnbd_iu *iu = priv;
 410         struct rnbd_clt_dev *dev = iu->dev;
 411         struct request *rq = iu->rq;
 412         int rw = rq_data_dir(rq);
 413
 414         iu->errno = errno;
 415
 416         blk_mq_complete_request(rq);
 417
 418         if (errno)
 419                 rnbd_clt_info_rl(dev, "%s I/O failed with err: %d\n",
 420                                  rw == READ ? "read" : "write", errno);
 421 }
 422
 423 static void wake_up_iu_comp(struct rnbd_iu *iu, int errno)
 424 {
 425         iu->comp.errno = errno;
 426         wake_up(&iu->comp.wait);
 427 }
 428
 429 static void msg_conf(void *priv, int errno)
 430 {
 431         struct rnbd_iu *iu = priv;
 432
 433         iu->errno = errno;
 434         schedule_work(&iu->work);
 435 }
 436
 437 static int send_usr_msg(struct rtrs_clt *rtrs, int dir,
 438                         struct rnbd_iu *iu, struct kvec *vec,
 439                         size_t len, struct scatterlist *sg, unsigned int sg_len,
 440                         void (*conf)(struct work_struct *work),
 441                         int *errno, int wait)
 442 {
 443         int err;
 444         struct rtrs_clt_req_ops req_ops;
 445
 446         INIT_WORK(&iu->work, conf);
 447         req_ops = (struct rtrs_clt_req_ops) {
 448                 .priv = iu,
 449                 .conf_fn = msg_conf,
 450         };
 451         err = rtrs_clt_request(dir, &req_ops, rtrs, iu->permit,
 452                                 vec, 1, len, sg, sg_len);
 453         if (!err && wait) {
 454                 wait_event(iu->comp.wait, iu->comp.errno != INT_MAX);
 455                 *errno = iu->comp.errno;
 456         } else {
 457                 *errno = 0;
 458         }
 459
 460         return err;
 461 }
 462
 463 static void msg_close_conf(struct work_struct *work)
 464 {
 465         struct rnbd_iu *iu = container_of(work, struct rnbd_iu, work);
 466         struct rnbd_clt_dev *dev = iu->dev;
 467
 468         wake_up_iu_comp(iu, iu->errno);
 469         rnbd_put_iu(dev->sess, iu);
 470         rnbd_clt_put_dev(dev);
 471 }
 472
 473 static int send_msg_close(struct rnbd_clt_dev *dev, u32 device_id,
 474                           enum wait_type wait)
 475 {
 476         struct rnbd_clt_session *sess = dev->sess;
 477         struct rnbd_msg_close msg;
 478         struct rnbd_iu *iu;
 479         struct kvec vec = {
 480                 .iov_base = &msg,
 481                 .iov_len  = sizeof(msg)
 482         };
 483         int err, errno;
 484
 485         iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
 486         if (!iu)
 487                 return -ENOMEM;
 488
 489         iu->buf = NULL;
 490         iu->dev = dev;
 491
 492         msg.hdr.type    = cpu_to_le16(RNBD_MSG_CLOSE);
 493         msg.device_id   = cpu_to_le32(device_id);
 494
 495         WARN_ON(!rnbd_clt_get_dev(dev));
 496         err = send_usr_msg(sess->rtrs, WRITE, iu, &vec, 0, NULL, 0,
 497                            msg_close_conf, &errno, wait);
 498         if (err) {
 499                 rnbd_clt_put_dev(dev);
 500                 rnbd_put_iu(sess, iu);
 501         } else {
 502                 err = errno;
 503         }
 504
 505         rnbd_put_iu(sess, iu);
 506         return err;
 507 }
 508
 509 static void msg_open_conf(struct work_struct *work)
 510 {
 511         struct rnbd_iu *iu = container_of(work, struct rnbd_iu, work);
 512         struct rnbd_msg_open_rsp *rsp = iu->buf;
 513         struct rnbd_clt_dev *dev = iu->dev;
 514         int errno = iu->errno;
 515
 516         if (errno) {
 517                 rnbd_clt_err(dev,
 518                               "Opening failed, server responded: %d\n",
 519                               errno);
 520         } else {
 521                 errno = process_msg_open_rsp(dev, rsp);
 522                 if (errno) {
 523                         u32 device_id = le32_to_cpu(rsp->device_id);
 524                         /*
 525                          * If server thinks its fine, but we fail to process
 526                          * then be nice and send a close to server.
 527                          */
 528                         send_msg_close(dev, device_id, RTRS_PERMIT_NOWAIT);
 529                 }
 530         }
 531         kfree(rsp);
 532         wake_up_iu_comp(iu, errno);
 533         rnbd_put_iu(dev->sess, iu);
 534         rnbd_clt_put_dev(dev);
 535 }
 536
 537 static void msg_sess_info_conf(struct work_struct *work)
 538 {
 539         struct rnbd_iu *iu = container_of(work, struct rnbd_iu, work);
 540         struct rnbd_msg_sess_info_rsp *rsp = iu->buf;
 541         struct rnbd_clt_session *sess = iu->sess;
 542
 543         if (!iu->errno)
 544                 sess->ver = min_t(u8, rsp->ver, RNBD_PROTO_VER_MAJOR);
 545
 546         kfree(rsp);
 547         wake_up_iu_comp(iu, iu->errno);
 548         rnbd_put_iu(sess, iu);
 549         rnbd_clt_put_sess(sess);
 550 }
 551
 552 static int send_msg_open(struct rnbd_clt_dev *dev, enum wait_type wait)
 553 {
 554         struct rnbd_clt_session *sess = dev->sess;
 555         struct rnbd_msg_open_rsp *rsp;
 556         struct rnbd_msg_open msg;
 557         struct rnbd_iu *iu;
 558         struct kvec vec = {
 559                 .iov_base = &msg,
 560                 .iov_len  = sizeof(msg)
 561         };
 562         int err, errno;
 563
 564         rsp = kzalloc(sizeof(*rsp), GFP_KERNEL);
 565         if (!rsp)
 566                 return -ENOMEM;
 567
 568         iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
 569         if (!iu) {
 570                 kfree(rsp);
 571                 return -ENOMEM;
 572         }
 573
 574         iu->buf = rsp;
 575         iu->dev = dev;
 576
 577         sg_init_one(iu->sgt.sgl, rsp, sizeof(*rsp));
 578
 579         msg.hdr.type    = cpu_to_le16(RNBD_MSG_OPEN);
 580         msg.access_mode = dev->access_mode;
 581         strscpy(msg.dev_name, dev->pathname, sizeof(msg.dev_name));
 582
 583         WARN_ON(!rnbd_clt_get_dev(dev));
 584         err = send_usr_msg(sess->rtrs, READ, iu,
 585                            &vec, sizeof(*rsp), iu->sgt.sgl, 1,
 586                            msg_open_conf, &errno, wait);
 587         if (err) {
 588                 rnbd_clt_put_dev(dev);
 589                 rnbd_put_iu(sess, iu);
 590                 kfree(rsp);
 591         } else {
 592                 err = errno;
 593         }
 594
 595         rnbd_put_iu(sess, iu);
 596         return err;
 597 }
 598
 599 static int send_msg_sess_info(struct rnbd_clt_session *sess, enum wait_type wait)
 600 {
 601         struct rnbd_msg_sess_info_rsp *rsp;
 602         struct rnbd_msg_sess_info msg;
 603         struct rnbd_iu *iu;
 604         struct kvec vec = {
 605                 .iov_base = &msg,
 606                 .iov_len  = sizeof(msg)
 607         };
 608         int err, errno;
 609
 610         rsp = kzalloc(sizeof(*rsp), GFP_KERNEL);
 611         if (!rsp)
 612                 return -ENOMEM;
 613
 614         iu = rnbd_get_iu(sess, RTRS_ADMIN_CON, RTRS_PERMIT_WAIT);
 615         if (!iu) {
 616                 kfree(rsp);
 617                 return -ENOMEM;
 618         }
 619
 620         iu->buf = rsp;
 621         iu->sess = sess;
 622         sg_init_one(iu->sgt.sgl, rsp, sizeof(*rsp));
 623
 624         msg.hdr.type = cpu_to_le16(RNBD_MSG_SESS_INFO);
 625         msg.ver      = RNBD_PROTO_VER_MAJOR;
 626
 627         if (!rnbd_clt_get_sess(sess)) {
 628                 /*
 629                  * That can happen only in one case, when RTRS has restablished
 630                  * the connection and link_ev() is called, but session is almost
 631                  * dead, last reference on session is put and caller is waiting
 632                  * for RTRS to close everything.
 633                  */
 634                 err = -ENODEV;
 635                 goto put_iu;
 636         }
 637         err = send_usr_msg(sess->rtrs, READ, iu,
 638                            &vec, sizeof(*rsp), iu->sgt.sgl, 1,
 639                            msg_sess_info_conf, &errno, wait);
 640         if (err) {
 641                 rnbd_clt_put_sess(sess);
 642 put_iu:
 643                 rnbd_put_iu(sess, iu);
 644                 kfree(rsp);
 645         } else {
 646                 err = errno;
 647         }
 648         rnbd_put_iu(sess, iu);
 649         return err;
 650 }
 651
 652 static void set_dev_states_to_disconnected(struct rnbd_clt_session *sess)
 653 {
 654         struct rnbd_clt_dev *dev;
 655         struct kobject *gd_kobj;
 656
 657         mutex_lock(&sess->lock);
 658         list_for_each_entry(dev, &sess->devs_list, list) {
 659                 rnbd_clt_err(dev, "Device disconnected.\n");
 660
 661                 mutex_lock(&dev->lock);
 662                 if (dev->dev_state == DEV_STATE_MAPPED) {
 663                         dev->dev_state = DEV_STATE_MAPPED_DISCONNECTED;
 664                         gd_kobj = &disk_to_dev(dev->gd)->kobj;
 665                         kobject_uevent(gd_kobj, KOBJ_OFFLINE);
 666                 }
 667                 mutex_unlock(&dev->lock);
 668         }
 669         mutex_unlock(&sess->lock);
 670 }
 671
 672 static void remap_devs(struct rnbd_clt_session *sess)
 673 {
 674         struct rnbd_clt_dev *dev;
 675         struct rtrs_attrs attrs;
 676         int err;
 677
 678         /*
 679          * Careful here: we are called from RTRS link event directly,
 680          * thus we can't send any RTRS request and wait for response
 681          * or RTRS will not be able to complete request with failure
 682          * if something goes wrong (failing of outstanding requests
 683          * happens exactly from the context where we are blocking now).
 684          *
 685          * So to avoid deadlocks each usr message sent from here must
 686          * be asynchronous.
 687          */
 688
 689         err = send_msg_sess_info(sess, RTRS_PERMIT_NOWAIT);
 690         if (err) {
 691                 pr_err("send_msg_sess_info(\"%s\"): %d\n", sess->sessname, err);
 692                 return;
 693         }
 694
 695         rtrs_clt_query(sess->rtrs, &attrs);
 696         mutex_lock(&sess->lock);
 697         sess->max_io_size = attrs.max_io_size;
 698
 699         list_for_each_entry(dev, &sess->devs_list, list) {
 700                 bool skip;
 701
 702                 mutex_lock(&dev->lock);
 703                 skip = (dev->dev_state == DEV_STATE_INIT);
 704                 mutex_unlock(&dev->lock);
 705                 if (skip)
 706                         /*
 707                          * When device is establishing connection for the first
 708                          * time - do not remap, it will be closed soon.
 709                          */
 710                         continue;
 711
 712                 rnbd_clt_info(dev, "session reconnected, remapping device\n");
 713                 err = send_msg_open(dev, RTRS_PERMIT_NOWAIT);
 714                 if (err) {
 715                         rnbd_clt_err(dev, "send_msg_open(): %d\n", err);
 716                         break;
 717                 }
 718         }
 719         mutex_unlock(&sess->lock);
 720 }
 721
 722 static void rnbd_clt_link_ev(void *priv, enum rtrs_clt_link_ev ev)
 723 {
 724         struct rnbd_clt_session *sess = priv;
 725
 726         switch (ev) {
 727         case RTRS_CLT_LINK_EV_DISCONNECTED:
 728                 set_dev_states_to_disconnected(sess);
 729                 break;
 730         case RTRS_CLT_LINK_EV_RECONNECTED:
 731                 remap_devs(sess);
 732                 break;
 733         default:
 734                 pr_err("Unknown session event received (%d), session: %s\n",
 735                        ev, sess->sessname);
 736         }
 737 }
 738
 739 static void rnbd_init_cpu_qlists(struct rnbd_cpu_qlist __percpu *cpu_queues)
 740 {
 741         unsigned int cpu;
 742         struct rnbd_cpu_qlist *cpu_q;
 743
 744         for_each_possible_cpu(cpu) {
 745                 cpu_q = per_cpu_ptr(cpu_queues, cpu);
 746
 747                 cpu_q->cpu = cpu;
 748                 INIT_LIST_HEAD(&cpu_q->requeue_list);
 749                 spin_lock_init(&cpu_q->requeue_lock);
 750         }
 751 }
 752
 753 static void destroy_mq_tags(struct rnbd_clt_session *sess)
 754 {
 755         if (sess->tag_set.tags)
 756                 blk_mq_free_tag_set(&sess->tag_set);
 757 }
 758
 759 static inline void wake_up_rtrs_waiters(struct rnbd_clt_session *sess)
 760 {
 761         sess->rtrs_ready = true;
 762         wake_up_all(&sess->rtrs_waitq);
 763 }
 764
 765 static void close_rtrs(struct rnbd_clt_session *sess)
 766 {
 767         might_sleep();
 768
 769         if (!IS_ERR_OR_NULL(sess->rtrs)) {
 770                 rtrs_clt_close(sess->rtrs);
 771                 sess->rtrs = NULL;
 772                 wake_up_rtrs_waiters(sess);
 773         }
 774 }
 775
 776 static void free_sess(struct rnbd_clt_session *sess)
 777 {
 778         WARN_ON(!list_empty(&sess->devs_list));
 779
 780         might_sleep();
 781
 782         close_rtrs(sess);
 783         destroy_mq_tags(sess);
 784         if (!list_empty(&sess->list)) {
 785                 mutex_lock(&sess_lock);
 786                 list_del(&sess->list);
 787                 mutex_unlock(&sess_lock);
 788         }
 789         free_percpu(sess->cpu_queues);
 790         free_percpu(sess->cpu_rr);
 791         mutex_destroy(&sess->lock);
 792         kfree(sess);
 793 }
 794
 795 static struct rnbd_clt_session *alloc_sess(const char *sessname)
 796 {
 797         struct rnbd_clt_session *sess;
 798         int err, cpu;
 799
 800         sess = kzalloc_node(sizeof(*sess), GFP_KERNEL, NUMA_NO_NODE);
 801         if (!sess)
 802                 return ERR_PTR(-ENOMEM);
 803         strscpy(sess->sessname, sessname, sizeof(sess->sessname));
 804         atomic_set(&sess->busy, 0);
 805         mutex_init(&sess->lock);
 806         INIT_LIST_HEAD(&sess->devs_list);
 807         INIT_LIST_HEAD(&sess->list);
 808         bitmap_zero(sess->cpu_queues_bm, NR_CPUS);
 809         init_waitqueue_head(&sess->rtrs_waitq);
 810         refcount_set(&sess->refcount, 1);
 811
 812         sess->cpu_queues = alloc_percpu(struct rnbd_cpu_qlist);
 813         if (!sess->cpu_queues) {
 814                 err = -ENOMEM;
 815                 goto err;
 816         }
 817         rnbd_init_cpu_qlists(sess->cpu_queues);
 818
 819         /*
 820          * That is simple percpu variable which stores cpu indices, which are
 821          * incremented on each access.  We need that for the sake of fairness
 822          * to wake up queues in a round-robin manner.
 823          */
 824         sess->cpu_rr = alloc_percpu(int);
 825         if (!sess->cpu_rr) {
 826                 err = -ENOMEM;
 827                 goto err;
 828         }
 829         for_each_possible_cpu(cpu)
 830                 * per_cpu_ptr(sess->cpu_rr, cpu) = cpu;
 831
 832         return sess;
 833
 834 err:
 835         free_sess(sess);
 836
 837         return ERR_PTR(err);
 838 }
 839
 840 static int wait_for_rtrs_connection(struct rnbd_clt_session *sess)
 841 {
 842         wait_event(sess->rtrs_waitq, sess->rtrs_ready);
 843         if (IS_ERR_OR_NULL(sess->rtrs))
 844                 return -ECONNRESET;
 845
 846         return 0;
 847 }
 848
 849 static void wait_for_rtrs_disconnection(struct rnbd_clt_session *sess)
 850         __releases(&sess_lock)
 851         __acquires(&sess_lock)
 852 {
 853         DEFINE_WAIT(wait);
 854
 855         prepare_to_wait(&sess->rtrs_waitq, &wait, TASK_UNINTERRUPTIBLE);
 856         if (IS_ERR_OR_NULL(sess->rtrs)) {
 857                 finish_wait(&sess->rtrs_waitq, &wait);
 858                 return;
 859         }
 860         mutex_unlock(&sess_lock);
 861         /* loop in caller, see __find_and_get_sess().
 862          * You can't leave mutex locked and call schedule(), you will catch a
 863          * deadlock with a caller of free_sess(), which has just put the last
 864          * reference and is about to take the sess_lock in order to delete
 865          * the session from the list.
 866          */
 867         schedule();
 868         mutex_lock(&sess_lock);
 869 }
 870
 871 static struct rnbd_clt_session *__find_and_get_sess(const char *sessname)
 872         __releases(&sess_lock)
 873         __acquires(&sess_lock)
 874 {
 875         struct rnbd_clt_session *sess, *sn;
 876         int err;
 877
 878 again:
 879         list_for_each_entry_safe(sess, sn, &sess_list, list) {
 880                 if (strcmp(sessname, sess->sessname))
 881                         continue;
 882
 883                 if (sess->rtrs_ready && IS_ERR_OR_NULL(sess->rtrs))
 884                         /*
 885                          * No RTRS connection, session is dying.
 886                          */
 887                         continue;
 888
 889                 if (rnbd_clt_get_sess(sess)) {
 890                         /*
 891                          * Alive session is found, wait for RTRS connection.
 892                          */
 893                         mutex_unlock(&sess_lock);
 894                         err = wait_for_rtrs_connection(sess);
 895                         if (err)
 896                                 rnbd_clt_put_sess(sess);
 897                         mutex_lock(&sess_lock);
 898
 899                         if (err)
 900                                 /* Session is dying, repeat the loop */
 901                                 goto again;
 902
 903                         return sess;
 904                 }
 905                 /*
 906                  * Ref is 0, session is dying, wait for RTRS disconnect
 907                  * in order to avoid session names clashes.
 908                  */
 909                 wait_for_rtrs_disconnection(sess);
 910                 /*
 911                  * RTRS is disconnected and soon session will be freed,
 912                  * so repeat a loop.
 913                  */
 914                 goto again;
 915         }
 916
 917         return NULL;
 918 }
 919
 920 /* caller is responsible for initializing 'first' to false */
 921 static struct
 922 rnbd_clt_session *find_or_create_sess(const char *sessname, bool *first)
 923 {
 924         struct rnbd_clt_session *sess = NULL;
 925
 926         mutex_lock(&sess_lock);
 927         sess = __find_and_get_sess(sessname);
 928         if (!sess) {
 929                 sess = alloc_sess(sessname);
 930                 if (IS_ERR(sess)) {
 931                         mutex_unlock(&sess_lock);
 932                         return sess;
 933                 }
 934                 list_add(&sess->list, &sess_list);
 935                 *first = true;
 936         }
 937         mutex_unlock(&sess_lock);
 938
 939         return sess;
 940 }
 941
 942 static int rnbd_client_open(struct block_device *block_device, fmode_t mode)
 943 {
 944         struct rnbd_clt_dev *dev = block_device->bd_disk->private_data;
 945
 946         if (dev->read_only && (mode & FMODE_WRITE))
 947                 return -EPERM;
 948
 949         if (dev->dev_state == DEV_STATE_UNMAPPED ||
 950             !rnbd_clt_get_dev(dev))
 951                 return -EIO;
 952
 953         return 0;
 954 }
 955
 956 static void rnbd_client_release(struct gendisk *gen, fmode_t mode)
 957 {
 958         struct rnbd_clt_dev *dev = gen->private_data;
 959
 960         rnbd_clt_put_dev(dev);
 961 }
 962
 963 static int rnbd_client_getgeo(struct block_device *block_device,
 964                               struct hd_geometry *geo)
 965 {
 966         u64 size;
 967         struct rnbd_clt_dev *dev;
 968
 969         dev = block_device->bd_disk->private_data;
 970         size = dev->size * (dev->logical_block_size / SECTOR_SIZE);
 971         geo->cylinders  = size >> 6;    /* size/64 */
 972         geo->heads      = 4;
 973         geo->sectors    = 16;
 974         geo->start      = 0;
 975
 976         return 0;
 977 }
 978
 979 static const struct block_device_operations rnbd_client_ops = {
 980         .owner          = THIS_MODULE,
 981         .open           = rnbd_client_open,
 982         .release        = rnbd_client_release,
 983         .getgeo         = rnbd_client_getgeo
 984 };
 985
 986 /* The amount of data that belongs to an I/O and the amount of data that
 987  * should be read or written to the disk (bi_size) can differ.
 988  *
 989  * E.g. When WRITE_SAME is used, only a small amount of data is
 990  * transferred that is then written repeatedly over a lot of sectors.
 991  *
 992  * Get the size of data to be transferred via RTRS by summing up the size
 993  * of the scather-gather list entries.
 994  */
 995 static size_t rnbd_clt_get_sg_size(struct scatterlist *sglist, u32 len)
 996 {
 997         struct scatterlist *sg;
 998         size_t tsize = 0;
 999         int i;
1000
1001         for_each_sg(sglist, sg, len, i)
1002                 tsize += sg->length;
1003         return tsize;
1004 }
1005
1006 static int rnbd_client_xfer_request(struct rnbd_clt_dev *dev,
1007                                      struct request *rq,
1008                                      struct rnbd_iu *iu)
1009 {
1010         struct rtrs_clt *rtrs = dev->sess->rtrs;
1011         struct rtrs_permit *permit = iu->permit;
1012         struct rnbd_msg_io msg;
1013         struct rtrs_clt_req_ops req_ops;
1014         unsigned int sg_cnt = 0;
1015         struct kvec vec;
1016         size_t size;
1017         int err;
1018
1019         iu->rq          = rq;
1020         iu->dev         = dev;
1021         msg.sector      = cpu_to_le64(blk_rq_pos(rq));
1022         msg.bi_size     = cpu_to_le32(blk_rq_bytes(rq));
1023         msg.rw          = cpu_to_le32(rq_to_rnbd_flags(rq));
1024         msg.prio        = cpu_to_le16(req_get_ioprio(rq));
1025
1026         /*
1027          * We only support discards with single segment for now.
1028          * See queue limits.
1029          */
1030         if (req_op(rq) != REQ_OP_DISCARD)
1031                 sg_cnt = blk_rq_map_sg(dev->queue, rq, iu->sgt.sgl);
1032
1033         if (sg_cnt == 0)
1034                 sg_mark_end(&iu->sgt.sgl[0]);
1035
1036         msg.hdr.type    = cpu_to_le16(RNBD_MSG_IO);
1037         msg.device_id   = cpu_to_le32(dev->device_id);
1038
1039         vec = (struct kvec) {
1040                 .iov_base = &msg,
1041                 .iov_len  = sizeof(msg)
1042         };
1043         size = rnbd_clt_get_sg_size(iu->sgt.sgl, sg_cnt);
1044         req_ops = (struct rtrs_clt_req_ops) {
1045                 .priv = iu,
1046                 .conf_fn = msg_io_conf,
1047         };
1048         err = rtrs_clt_request(rq_data_dir(rq), &req_ops, rtrs, permit,
1049                                &vec, 1, size, iu->sgt.sgl, sg_cnt);
1050         if (unlikely(err)) {
1051                 rnbd_clt_err_rl(dev, "RTRS failed to transfer IO, err: %d\n",
1052                                  err);
1053                 return err;
1054         }
1055
1056         return 0;
1057 }
1058
1059 /**
1060  * rnbd_clt_dev_add_to_requeue() - add device to requeue if session is busy
1061  * @dev:        Device to be checked
1062  * @q:          Queue to be added to the requeue list if required
1063  *
1064  * Description:
1065  *     If session is busy, that means someone will requeue us when resources
1066  *     are freed.  If session is not doing anything - device is not added to
1067  *     the list and @false is returned.
1068  */
1069 static bool rnbd_clt_dev_add_to_requeue(struct rnbd_clt_dev *dev,
1070                                                 struct rnbd_queue *q)
1071 {
1072         struct rnbd_clt_session *sess = dev->sess;
1073         struct rnbd_cpu_qlist *cpu_q;
1074         unsigned long flags;
1075         bool added = true;
1076         bool need_set;
1077
1078         cpu_q = get_cpu_ptr(sess->cpu_queues);
1079         spin_lock_irqsave(&cpu_q->requeue_lock, flags);
1080
1081         if (likely(!test_and_set_bit_lock(0, &q->in_list))) {
1082                 if (WARN_ON(!list_empty(&q->requeue_list)))
1083                         goto unlock;
1084
1085                 need_set = !test_bit(cpu_q->cpu, sess->cpu_queues_bm);
1086                 if (need_set) {
1087                         set_bit(cpu_q->cpu, sess->cpu_queues_bm);
1088                         /* Paired with rnbd_put_permit(). Set a bit first
1089                          * and then observe the busy counter.
1090                          */
1091                         smp_mb__before_atomic();
1092                 }
1093                 if (likely(atomic_read(&sess->busy))) {
1094                         list_add_tail(&q->requeue_list, &cpu_q->requeue_list);
1095                 } else {
1096                         /* Very unlikely, but possible: busy counter was
1097                          * observed as zero.  Drop all bits and return
1098                          * false to restart the queue by ourselves.
1099                          */
1100                         if (need_set)
1101                                 clear_bit(cpu_q->cpu, sess->cpu_queues_bm);
1102                         clear_bit_unlock(0, &q->in_list);
1103                         added = false;
1104                 }
1105         }
1106 unlock:
1107         spin_unlock_irqrestore(&cpu_q->requeue_lock, flags);
1108         put_cpu_ptr(sess->cpu_queues);
1109
1110         return added;
1111 }
1112
1113 static void rnbd_clt_dev_kick_mq_queue(struct rnbd_clt_dev *dev,
1114                                         struct blk_mq_hw_ctx *hctx,
1115                                         int delay)
1116 {
1117         struct rnbd_queue *q = hctx->driver_data;
1118
1119         if (delay != RNBD_DELAY_IFBUSY)
1120                 blk_mq_delay_run_hw_queue(hctx, delay);
1121         else if (unlikely(!rnbd_clt_dev_add_to_requeue(dev, q)))
1122                 /*
1123                  * If session is not busy we have to restart
1124                  * the queue ourselves.
1125                  */
1126                 blk_mq_delay_run_hw_queue(hctx, 10/*ms*/);
1127 }
1128
1129 static blk_status_t rnbd_queue_rq(struct blk_mq_hw_ctx *hctx,
1130                                    const struct blk_mq_queue_data *bd)
1131 {
1132         struct request *rq = bd->rq;
1133         struct rnbd_clt_dev *dev = rq->rq_disk->private_data;
1134         struct rnbd_iu *iu = blk_mq_rq_to_pdu(rq);
1135         int err;
1136         blk_status_t ret = BLK_STS_IOERR;
1137
1138         if (unlikely(dev->dev_state != DEV_STATE_MAPPED))
1139                 return BLK_STS_IOERR;
1140
1141         iu->permit = rnbd_get_permit(dev->sess, RTRS_IO_CON,
1142                                       RTRS_PERMIT_NOWAIT);
1143         if (unlikely(!iu->permit)) {
1144                 rnbd_clt_dev_kick_mq_queue(dev, hctx, RNBD_DELAY_IFBUSY);
1145                 return BLK_STS_RESOURCE;
1146         }
1147
1148         iu->sgt.sgl = iu->first_sgl;
1149         err = sg_alloc_table_chained(&iu->sgt,
1150                                      /* Even-if the request has no segment,
1151                                       * sglist must have one entry at least */
1152                                      blk_rq_nr_phys_segments(rq) ? : 1,
1153                                      iu->sgt.sgl,
1154                                      RNBD_INLINE_SG_CNT);
1155         if (err) {
1156                 rnbd_clt_err_rl(dev, "sg_alloc_table_chained ret=%d\n", err);
1157                 rnbd_clt_dev_kick_mq_queue(dev, hctx, 10/*ms*/);
1158                 rnbd_put_permit(dev->sess, iu->permit);
1159                 return BLK_STS_RESOURCE;
1160         }
1161
1162         blk_mq_start_request(rq);
1163         err = rnbd_client_xfer_request(dev, rq, iu);
1164         if (likely(err == 0))
1165                 return BLK_STS_OK;
1166         if (unlikely(err == -EAGAIN || err == -ENOMEM)) {
1167                 rnbd_clt_dev_kick_mq_queue(dev, hctx, 10/*ms*/);
1168                 ret = BLK_STS_RESOURCE;
1169         }
1170         sg_free_table_chained(&iu->sgt, RNBD_INLINE_SG_CNT);
1171         rnbd_put_permit(dev->sess, iu->permit);
1172         return ret;
1173 }
1174
1175 static int rnbd_rdma_poll(struct blk_mq_hw_ctx *hctx)
1176 {
1177         struct rnbd_queue *q = hctx->driver_data;
1178         struct rnbd_clt_dev *dev = q->dev;
1179         int cnt;
1180
1181         cnt = rtrs_clt_rdma_cq_direct(dev->sess->rtrs, hctx->queue_num);
1182         return cnt;
1183 }
1184
1185 static int rnbd_rdma_map_queues(struct blk_mq_tag_set *set)
1186 {
1187         struct rnbd_clt_session *sess = set->driver_data;
1188
1189         /* shared read/write queues */
1190         set->map[HCTX_TYPE_DEFAULT].nr_queues = num_online_cpus();
1191         set->map[HCTX_TYPE_DEFAULT].queue_offset = 0;
1192         set->map[HCTX_TYPE_READ].nr_queues = num_online_cpus();
1193         set->map[HCTX_TYPE_READ].queue_offset = 0;
1194         blk_mq_map_queues(&set->map[HCTX_TYPE_DEFAULT]);
1195         blk_mq_map_queues(&set->map[HCTX_TYPE_READ]);
1196
1197         if (sess->nr_poll_queues) {
1198                 /* dedicated queue for poll */
1199                 set->map[HCTX_TYPE_POLL].nr_queues = sess->nr_poll_queues;
1200                 set->map[HCTX_TYPE_POLL].queue_offset = set->map[HCTX_TYPE_READ].queue_offset +
1201                         set->map[HCTX_TYPE_READ].nr_queues;
1202                 blk_mq_map_queues(&set->map[HCTX_TYPE_POLL]);
1203                 pr_info("[session=%s] mapped %d/%d/%d default/read/poll queues.\n",
1204                         sess->sessname,
1205                         set->map[HCTX_TYPE_DEFAULT].nr_queues,
1206                         set->map[HCTX_TYPE_READ].nr_queues,
1207                         set->map[HCTX_TYPE_POLL].nr_queues);
1208         } else {
1209                 pr_info("[session=%s] mapped %d/%d default/read queues.\n",
1210                         sess->sessname,
1211                         set->map[HCTX_TYPE_DEFAULT].nr_queues,
1212                         set->map[HCTX_TYPE_READ].nr_queues);
1213         }
1214
1215         return 0;
1216 }
1217
1218 static struct blk_mq_ops rnbd_mq_ops = {
1219         .queue_rq       = rnbd_queue_rq,
1220         .complete       = rnbd_softirq_done_fn,
1221         .map_queues     = rnbd_rdma_map_queues,
1222         .poll           = rnbd_rdma_poll,
1223 };
1224
1225 static int setup_mq_tags(struct rnbd_clt_session *sess)
1226 {
1227         struct blk_mq_tag_set *tag_set = &sess->tag_set;
1228
1229         memset(tag_set, 0, sizeof(*tag_set));
1230         tag_set->ops            = &rnbd_mq_ops;
1231         tag_set->queue_depth    = sess->queue_depth;
1232         tag_set->numa_node              = NUMA_NO_NODE;
1233         tag_set->flags          = BLK_MQ_F_SHOULD_MERGE |
1234                                   BLK_MQ_F_TAG_QUEUE_SHARED;
1235         tag_set->cmd_size       = sizeof(struct rnbd_iu) + RNBD_RDMA_SGL_SIZE;
1236
1237         /* for HCTX_TYPE_DEFAULT, HCTX_TYPE_READ, HCTX_TYPE_POLL */
1238         tag_set->nr_maps        = sess->nr_poll_queues ? HCTX_MAX_TYPES : 2;
1239         /*
1240          * HCTX_TYPE_DEFAULT and HCTX_TYPE_READ share one set of queues
1241          * others are for HCTX_TYPE_POLL
1242          */
1243         tag_set->nr_hw_queues   = num_online_cpus() + sess->nr_poll_queues;
1244         tag_set->driver_data    = sess;
1245
1246         return blk_mq_alloc_tag_set(tag_set);
1247 }
1248
1249 static struct rnbd_clt_session *
1250 find_and_get_or_create_sess(const char *sessname,
1251                             const struct rtrs_addr *paths,
1252                             size_t path_cnt, u16 port_nr, u32 nr_poll_queues)
1253 {
1254         struct rnbd_clt_session *sess;
1255         struct rtrs_attrs attrs;
1256         int err;
1257         bool first = false;
1258         struct rtrs_clt_ops rtrs_ops;
1259
1260         sess = find_or_create_sess(sessname, &first);
1261         if (sess == ERR_PTR(-ENOMEM))
1262                 return ERR_PTR(-ENOMEM);
1263         else if ((nr_poll_queues && !first) ||  (!nr_poll_queues && sess->nr_poll_queues)) {
1264                 /*
1265                  * A device MUST have its own session to use the polling-mode.
1266                  * It must fail to map new device with the same session.
1267                  */
1268                 err = -EINVAL;
1269                 goto put_sess;
1270         }
1271
1272         if (!first)
1273                 return sess;
1274
1275         if (!path_cnt) {
1276                 pr_err("Session %s not found, and path parameter not given", sessname);
1277                 err = -ENXIO;
1278                 goto put_sess;
1279         }
1280
1281         rtrs_ops = (struct rtrs_clt_ops) {
1282                 .priv = sess,
1283                 .link_ev = rnbd_clt_link_ev,
1284         };
1285         /*
1286          * Nothing was found, establish rtrs connection and proceed further.
1287          */
1288         sess->rtrs = rtrs_clt_open(&rtrs_ops, sessname,
1289                                    paths, path_cnt, port_nr,
1290                                    0, /* Do not use pdu of rtrs */
1291                                    RECONNECT_DELAY, BMAX_SEGMENTS,
1292                                    MAX_RECONNECTS, nr_poll_queues);
1293         if (IS_ERR(sess->rtrs)) {
1294                 err = PTR_ERR(sess->rtrs);
1295                 goto wake_up_and_put;
1296         }
1297         rtrs_clt_query(sess->rtrs, &attrs);
1298         sess->max_io_size = attrs.max_io_size;
1299         sess->queue_depth = attrs.queue_depth;
1300         sess->nr_poll_queues = nr_poll_queues;
1301
1302         err = setup_mq_tags(sess);
1303         if (err)
1304                 goto close_rtrs;
1305
1306         err = send_msg_sess_info(sess, RTRS_PERMIT_WAIT);
1307         if (err)
1308                 goto close_rtrs;
1309
1310         wake_up_rtrs_waiters(sess);
1311
1312         return sess;
1313
1314 close_rtrs:
1315         close_rtrs(sess);
1316 put_sess:
1317         rnbd_clt_put_sess(sess);
1318
1319         return ERR_PTR(err);
1320
1321 wake_up_and_put:
1322         wake_up_rtrs_waiters(sess);
1323         goto put_sess;
1324 }
1325
1326 static inline void rnbd_init_hw_queue(struct rnbd_clt_dev *dev,
1327                                        struct rnbd_queue *q,
1328                                        struct blk_mq_hw_ctx *hctx)
1329 {
1330         INIT_LIST_HEAD(&q->requeue_list);
1331         q->dev  = dev;
1332         q->hctx = hctx;
1333 }
1334
1335 static void rnbd_init_mq_hw_queues(struct rnbd_clt_dev *dev)
1336 {
1337         int i;
1338         struct blk_mq_hw_ctx *hctx;
1339         struct rnbd_queue *q;
1340
1341         queue_for_each_hw_ctx(dev->queue, hctx, i) {
1342                 q = &dev->hw_queues[i];
1343                 rnbd_init_hw_queue(dev, q, hctx);
1344                 hctx->driver_data = q;
1345         }
1346 }
1347
1348 static int setup_mq_dev(struct rnbd_clt_dev *dev)
1349 {
1350         dev->queue = blk_mq_init_queue(&dev->sess->tag_set);
1351         if (IS_ERR(dev->queue)) {
1352                 rnbd_clt_err(dev, "Initializing multiqueue queue failed, err: %ld\n",
1353                               PTR_ERR(dev->queue));
1354                 return PTR_ERR(dev->queue);
1355         }
1356         rnbd_init_mq_hw_queues(dev);
1357         return 0;
1358 }
1359
1360 static void setup_request_queue(struct rnbd_clt_dev *dev)
1361 {
1362         blk_queue_logical_block_size(dev->queue, dev->logical_block_size);
1363         blk_queue_physical_block_size(dev->queue, dev->physical_block_size);
1364         blk_queue_max_hw_sectors(dev->queue, dev->max_hw_sectors);
1365         blk_queue_max_write_same_sectors(dev->queue,
1366                                          dev->max_write_same_sectors);
1367
1368         /*
1369          * we don't support discards to "discontiguous" segments
1370          * in on request
1371          */
1372         blk_queue_max_discard_segments(dev->queue, 1);
1373
1374         blk_queue_max_discard_sectors(dev->queue, dev->max_discard_sectors);
1375         dev->queue->limits.discard_granularity  = dev->discard_granularity;
1376         dev->queue->limits.discard_alignment    = dev->discard_alignment;
1377         if (dev->max_discard_sectors)
1378                 blk_queue_flag_set(QUEUE_FLAG_DISCARD, dev->queue);
1379         if (dev->secure_discard)
1380                 blk_queue_flag_set(QUEUE_FLAG_SECERASE, dev->queue);
1381
1382         blk_queue_flag_set(QUEUE_FLAG_SAME_COMP, dev->queue);
1383         blk_queue_flag_set(QUEUE_FLAG_SAME_FORCE, dev->queue);
1384         blk_queue_max_segments(dev->queue, dev->max_segments);
1385         blk_queue_io_opt(dev->queue, dev->sess->max_io_size);
1386         blk_queue_virt_boundary(dev->queue, SZ_4K - 1);
1387         blk_queue_write_cache(dev->queue, dev->wc, dev->fua);
1388         dev->queue->queuedata = dev;
1389 }
1390
1391 static void rnbd_clt_setup_gen_disk(struct rnbd_clt_dev *dev, int idx)
1392 {
1393         dev->gd->major          = rnbd_client_major;
1394         dev->gd->first_minor    = idx << RNBD_PART_BITS;
1395         dev->gd->fops           = &rnbd_client_ops;
1396         dev->gd->queue          = dev->queue;
1397         dev->gd->private_data   = dev;
1398         snprintf(dev->gd->disk_name, sizeof(dev->gd->disk_name), "rnbd%d",
1399                  idx);
1400         pr_debug("disk_name=%s, capacity=%zu\n",
1401                  dev->gd->disk_name,
1402                  dev->nsectors * (dev->logical_block_size / SECTOR_SIZE)
1403                  );
1404
1405         set_capacity(dev->gd, dev->nsectors);
1406
1407         if (dev->access_mode == RNBD_ACCESS_RO) {
1408                 dev->read_only = true;
1409                 set_disk_ro(dev->gd, true);
1410         } else {
1411                 dev->read_only = false;
1412         }
1413
1414         if (!dev->rotational)
1415                 blk_queue_flag_set(QUEUE_FLAG_NONROT, dev->queue);
1416         add_disk(dev->gd);
1417 }
1418
1419 static int rnbd_client_setup_device(struct rnbd_clt_dev *dev)
1420 {
1421         int err, idx = dev->clt_device_id;
1422
1423         dev->size = dev->nsectors * dev->logical_block_size;
1424
1425         err = setup_mq_dev(dev);
1426         if (err)
1427                 return err;
1428
1429         setup_request_queue(dev);
1430
1431         dev->gd = alloc_disk_node(1 << RNBD_PART_BITS,  NUMA_NO_NODE);
1432         if (!dev->gd) {
1433                 blk_cleanup_queue(dev->queue);
1434                 return -ENOMEM;
1435         }
1436
1437         rnbd_clt_setup_gen_disk(dev, idx);
1438
1439         return 0;
1440 }
1441
1442 static struct rnbd_clt_dev *init_dev(struct rnbd_clt_session *sess,
1443                                       enum rnbd_access_mode access_mode,
1444                                       const char *pathname,
1445                                       u32 nr_poll_queues)
1446 {
1447         struct rnbd_clt_dev *dev;
1448         int ret;
1449
1450         dev = kzalloc_node(sizeof(*dev), GFP_KERNEL, NUMA_NO_NODE);
1451         if (!dev)
1452                 return ERR_PTR(-ENOMEM);
1453
1454         /*
1455          * nr_cpu_ids: the number of softirq queues
1456          * nr_poll_queues: the number of polling queues
1457          */
1458         dev->hw_queues = kcalloc(nr_cpu_ids + nr_poll_queues,
1459                                  sizeof(*dev->hw_queues),
1460                                  GFP_KERNEL);
1461         if (!dev->hw_queues) {
1462                 ret = -ENOMEM;
1463                 goto out_alloc;
1464         }
1465
1466         mutex_lock(&ida_lock);
1467         ret = ida_simple_get(&index_ida, 0, 1 << (MINORBITS - RNBD_PART_BITS),
1468                              GFP_KERNEL);
1469         mutex_unlock(&ida_lock);
1470         if (ret < 0) {
1471                 pr_err("Failed to initialize device '%s' from session %s, allocating idr failed, err: %d\n",
1472                        pathname, sess->sessname, ret);
1473                 goto out_queues;
1474         }
1475
1476         dev->pathname = kstrdup(pathname, GFP_KERNEL);
1477         if (!dev->pathname) {
1478                 ret = -ENOMEM;
1479                 goto out_queues;
1480         }
1481
1482         dev->clt_device_id      = ret;
1483         dev->sess               = sess;
1484         dev->access_mode        = access_mode;
1485         dev->nr_poll_queues     = nr_poll_queues;
1486         mutex_init(&dev->lock);
1487         refcount_set(&dev->refcount, 1);
1488         dev->dev_state = DEV_STATE_INIT;
1489
1490         /*
1491          * Here we called from sysfs entry, thus clt-sysfs is
1492          * responsible that session will not disappear.
1493          */
1494         WARN_ON(!rnbd_clt_get_sess(sess));
1495
1496         return dev;
1497
1498 out_queues:
1499         kfree(dev->hw_queues);
1500 out_alloc:
1501         kfree(dev);
1502         return ERR_PTR(ret);
1503 }
1504
1505 static bool __exists_dev(const char *pathname, const char *sessname)
1506 {
1507         struct rnbd_clt_session *sess;
1508         struct rnbd_clt_dev *dev;
1509         bool found = false;
1510
1511         list_for_each_entry(sess, &sess_list, list) {
1512                 if (sessname && strncmp(sess->sessname, sessname,
1513                                         sizeof(sess->sessname)))
1514                         continue;
1515                 mutex_lock(&sess->lock);
1516                 list_for_each_entry(dev, &sess->devs_list, list) {
1517                         if (strlen(dev->pathname) == strlen(pathname) &&
1518                             !strcmp(dev->pathname, pathname)) {
1519                                 found = true;
1520                                 break;
1521                         }
1522                 }
1523                 mutex_unlock(&sess->lock);
1524                 if (found)
1525                         break;
1526         }
1527
1528         return found;
1529 }
1530
1531 static bool exists_devpath(const char *pathname, const char *sessname)
1532 {
1533         bool found;
1534
1535         mutex_lock(&sess_lock);
1536         found = __exists_dev(pathname, sessname);
1537         mutex_unlock(&sess_lock);
1538
1539         return found;
1540 }
1541
1542 static bool insert_dev_if_not_exists_devpath(struct rnbd_clt_dev *dev)
1543 {
1544         bool found;
1545         struct rnbd_clt_session *sess = dev->sess;
1546
1547         mutex_lock(&sess_lock);
1548         found = __exists_dev(dev->pathname, sess->sessname);
1549         if (!found) {
1550                 mutex_lock(&sess->lock);
1551                 list_add_tail(&dev->list, &sess->devs_list);
1552                 mutex_unlock(&sess->lock);
1553         }
1554         mutex_unlock(&sess_lock);
1555
1556         return found;
1557 }
1558
1559 static void delete_dev(struct rnbd_clt_dev *dev)
1560 {
1561         struct rnbd_clt_session *sess = dev->sess;
1562
1563         mutex_lock(&sess->lock);
1564         list_del(&dev->list);
1565         mutex_unlock(&sess->lock);
1566 }
1567
1568 struct rnbd_clt_dev *rnbd_clt_map_device(const char *sessname,
1569                                            struct rtrs_addr *paths,
1570                                            size_t path_cnt, u16 port_nr,
1571                                            const char *pathname,
1572                                            enum rnbd_access_mode access_mode,
1573                                            u32 nr_poll_queues)
1574 {
1575         struct rnbd_clt_session *sess;
1576         struct rnbd_clt_dev *dev;
1577         int ret;
1578
1579         if (unlikely(exists_devpath(pathname, sessname)))
1580                 return ERR_PTR(-EEXIST);
1581
1582         sess = find_and_get_or_create_sess(sessname, paths, path_cnt, port_nr, nr_poll_queues);
1583         if (IS_ERR(sess))
1584                 return ERR_CAST(sess);
1585
1586         dev = init_dev(sess, access_mode, pathname, nr_poll_queues);
1587         if (IS_ERR(dev)) {
1588                 pr_err("map_device: failed to map device '%s' from session %s, can't initialize device, err: %ld\n",
1589                        pathname, sess->sessname, PTR_ERR(dev));
1590                 ret = PTR_ERR(dev);
1591                 goto put_sess;
1592         }
1593         if (insert_dev_if_not_exists_devpath(dev)) {
1594                 ret = -EEXIST;
1595                 goto put_dev;
1596         }
1597         ret = send_msg_open(dev, RTRS_PERMIT_WAIT);
1598         if (ret) {
1599                 rnbd_clt_err(dev,
1600                               "map_device: failed, can't open remote device, err: %d\n",
1601                               ret);
1602                 goto del_dev;
1603         }
1604         mutex_lock(&dev->lock);
1605         pr_debug("Opened remote device: session=%s, path='%s'\n",
1606                  sess->sessname, pathname);
1607         ret = rnbd_client_setup_device(dev);
1608         if (ret) {
1609                 rnbd_clt_err(dev,
1610                               "map_device: Failed to configure device, err: %d\n",
1611                               ret);
1612                 mutex_unlock(&dev->lock);
1613                 goto send_close;
1614         }
1615
1616         rnbd_clt_info(dev,
1617                        "map_device: Device mapped as %s (nsectors: %zu, logical_block_size: %d, physical_block_size: %d, max_write_same_sectors: %d, max_discard_sectors: %d, discard_granularity: %d, discard_alignment: %d, secure_discard: %d, max_segments: %d, max_hw_sectors: %d, rotational: %d, wc: %d, fua: %d)\n",
1618                        dev->gd->disk_name, dev->nsectors,
1619                        dev->logical_block_size, dev->physical_block_size,
1620                        dev->max_write_same_sectors, dev->max_discard_sectors,
1621                        dev->discard_granularity, dev->discard_alignment,
1622                        dev->secure_discard, dev->max_segments,
1623                        dev->max_hw_sectors, dev->rotational, dev->wc, dev->fua);
1624
1625         mutex_unlock(&dev->lock);
1626         rnbd_clt_put_sess(sess);
1627
1628         return dev;
1629
1630 send_close:
1631         send_msg_close(dev, dev->device_id, RTRS_PERMIT_WAIT);
1632 del_dev:
1633         delete_dev(dev);
1634 put_dev:
1635         rnbd_clt_put_dev(dev);
1636 put_sess:
1637         rnbd_clt_put_sess(sess);
1638
1639         return ERR_PTR(ret);
1640 }
1641
1642 static void destroy_gen_disk(struct rnbd_clt_dev *dev)
1643 {
1644         del_gendisk(dev->gd);
1645         blk_cleanup_queue(dev->queue);
1646         put_disk(dev->gd);
1647 }
1648
1649 static void destroy_sysfs(struct rnbd_clt_dev *dev,
1650                           const struct attribute *sysfs_self)
1651 {
1652         rnbd_clt_remove_dev_symlink(dev);
1653         if (dev->kobj.state_initialized) {
1654                 if (sysfs_self)
1655                         /* To avoid deadlock firstly remove itself */
1656                         sysfs_remove_file_self(&dev->kobj, sysfs_self);
1657                 kobject_del(&dev->kobj);
1658                 kobject_put(&dev->kobj);
1659         }
1660 }
1661
1662 int rnbd_clt_unmap_device(struct rnbd_clt_dev *dev, bool force,
1663                            const struct attribute *sysfs_self)
1664 {
1665         struct rnbd_clt_session *sess = dev->sess;
1666         int refcount, ret = 0;
1667         bool was_mapped;
1668
1669         mutex_lock(&dev->lock);
1670         if (dev->dev_state == DEV_STATE_UNMAPPED) {
1671                 rnbd_clt_info(dev, "Device is already being unmapped\n");
1672                 ret = -EALREADY;
1673                 goto err;
1674         }
1675         refcount = refcount_read(&dev->refcount);
1676         if (!force && refcount > 1) {
1677                 rnbd_clt_err(dev,
1678                               "Closing device failed, device is in use, (%d device users)\n",
1679                               refcount - 1);
1680                 ret = -EBUSY;
1681                 goto err;
1682         }
1683         was_mapped = (dev->dev_state == DEV_STATE_MAPPED);
1684         dev->dev_state = DEV_STATE_UNMAPPED;
1685         mutex_unlock(&dev->lock);
1686
1687         delete_dev(dev);
1688         destroy_sysfs(dev, sysfs_self);
1689         destroy_gen_disk(dev);
1690         if (was_mapped && sess->rtrs)
1691                 send_msg_close(dev, dev->device_id, RTRS_PERMIT_WAIT);
1692
1693         rnbd_clt_info(dev, "Device is unmapped\n");
1694
1695         /* Likely last reference put */
1696         rnbd_clt_put_dev(dev);
1697
1698         /*
1699          * Here device and session can be vanished!
1700          */
1701
1702         return 0;
1703 err:
1704         mutex_unlock(&dev->lock);
1705
1706         return ret;
1707 }
1708
1709 int rnbd_clt_remap_device(struct rnbd_clt_dev *dev)
1710 {
1711         int err;
1712
1713         mutex_lock(&dev->lock);
1714         if (dev->dev_state == DEV_STATE_MAPPED_DISCONNECTED)
1715                 err = 0;
1716         else if (dev->dev_state == DEV_STATE_UNMAPPED)
1717                 err = -ENODEV;
1718         else if (dev->dev_state == DEV_STATE_MAPPED)
1719                 err = -EALREADY;
1720         else
1721                 err = -EBUSY;
1722         mutex_unlock(&dev->lock);
1723         if (!err) {
1724                 rnbd_clt_info(dev, "Remapping device.\n");
1725                 err = send_msg_open(dev, RTRS_PERMIT_WAIT);
1726                 if (err)
1727                         rnbd_clt_err(dev, "remap_device: %d\n", err);
1728         }
1729
1730         return err;
1731 }
1732
1733 static void unmap_device_work(struct work_struct *work)
1734 {
1735         struct rnbd_clt_dev *dev;
1736
1737         dev = container_of(work, typeof(*dev), unmap_on_rmmod_work);
1738         rnbd_clt_unmap_device(dev, true, NULL);
1739 }
1740
1741 static void rnbd_destroy_sessions(void)
1742 {
1743         struct rnbd_clt_session *sess, *sn;
1744         struct rnbd_clt_dev *dev, *tn;
1745
1746         /* Firstly forbid access through sysfs interface */
1747         rnbd_clt_destroy_sysfs_files();
1748
1749         /*
1750          * Here at this point there is no any concurrent access to sessions
1751          * list and devices list:
1752          *   1. New session or device can't be created - session sysfs files
1753          *      are removed.
1754          *   2. Device or session can't be removed - module reference is taken
1755          *      into account in unmap device sysfs callback.
1756          *   3. No IO requests inflight - each file open of block_dev increases
1757          *      module reference in get_disk().
1758          *
1759          * But still there can be user requests inflights, which are sent by
1760          * asynchronous send_msg_*() functions, thus before unmapping devices
1761          * RTRS session must be explicitly closed.
1762          */
1763
1764         list_for_each_entry_safe(sess, sn, &sess_list, list) {
1765                 if (!rnbd_clt_get_sess(sess))
1766                         continue;
1767                 close_rtrs(sess);
1768                 list_for_each_entry_safe(dev, tn, &sess->devs_list, list) {
1769                         /*
1770                          * Here unmap happens in parallel for only one reason:
1771                          * blk_cleanup_queue() takes around half a second, so
1772                          * on huge amount of devices the whole module unload
1773                          * procedure takes minutes.
1774                          */
1775                         INIT_WORK(&dev->unmap_on_rmmod_work, unmap_device_work);
1776                         queue_work(system_long_wq, &dev->unmap_on_rmmod_work);
1777                 }
1778                 rnbd_clt_put_sess(sess);
1779         }
1780         /* Wait for all scheduled unmap works */
1781         flush_workqueue(system_long_wq);
1782         WARN_ON(!list_empty(&sess_list));
1783 }
1784
1785 static int __init rnbd_client_init(void)
1786 {
1787         int err = 0;
1788
1789         BUILD_BUG_ON(sizeof(struct rnbd_msg_hdr) != 4);
1790         BUILD_BUG_ON(sizeof(struct rnbd_msg_sess_info) != 36);
1791         BUILD_BUG_ON(sizeof(struct rnbd_msg_sess_info_rsp) != 36);
1792         BUILD_BUG_ON(sizeof(struct rnbd_msg_open) != 264);
1793         BUILD_BUG_ON(sizeof(struct rnbd_msg_close) != 8);
1794         BUILD_BUG_ON(sizeof(struct rnbd_msg_open_rsp) != 56);
1795         rnbd_client_major = register_blkdev(rnbd_client_major, "rnbd");
1796         if (rnbd_client_major <= 0) {
1797                 pr_err("Failed to load module, block device registration failed\n");
1798                 return -EBUSY;
1799         }
1800
1801         err = rnbd_clt_create_sysfs_files();
1802         if (err) {
1803                 pr_err("Failed to load module, creating sysfs device files failed, err: %d\n",
1804                        err);
1805                 unregister_blkdev(rnbd_client_major, "rnbd");
1806         }
1807
1808         return err;
1809 }
1810
1811 static void __exit rnbd_client_exit(void)
1812 {
1813         rnbd_destroy_sessions();
1814         unregister_blkdev(rnbd_client_major, "rnbd");
1815         ida_destroy(&index_ida);
1816 }
1817
1818 module_init(rnbd_client_init);
1819 module_exit(rnbd_client_exit);