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