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