sys/dev/disk/xdisk/xdisk.c

   1 /*
   2  * Copyright (c) 2012-2014 The DragonFly Project.  All rights reserved.
   3  *
   4  * This code is derived from software contributed to The DragonFly Project
   5  * by Matthew Dillon <dillon@dragonflybsd.org>
   6  *
   7  * Redistribution and use in source and binary forms, with or without
   8  * modification, are permitted provided that the following conditions
   9  * are met:
  10  *
  11  * 1. Redistributions of source code must retain the above copyright
  12  *    notice, this list of conditions and the following disclaimer.
  13  * 2. Redistributions in binary form must reproduce the above copyright
  14  *    notice, this list of conditions and the following disclaimer in
  15  *    the documentation and/or other materials provided with the
  16  *    distribution.
  17  * 3. Neither the name of The DragonFly Project nor the names of its
  18  *    contributors may be used to endorse or promote products derived
  19  *    from this software without specific, prior written permission.
  20  *
  21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
  24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
  25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
  26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
  27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  32  * SUCH DAMAGE.
  33  */
  34 /*
  35  * This module allows disk devices to be created and associated with a
  36  * communications pipe or socket.  You open the device and issue an
  37  * ioctl() to install a new disk along with its communications descriptor.
  38  *
  39  * All further communication occurs via the descriptor using the DMSG
  40  * LNK_CONN, LNK_SPAN, and BLOCK protocols.  The descriptor can be a
  41  * direct connection to a remote machine's disk (in-kernenl), to a remote
  42  * cluster controller, to the local cluster controller, etc.
  43  *
  44  * /dev/xdisk is the control device, issue ioctl()s to create the /dev/xa%d
  45  * devices.  These devices look like raw disks to the system.
  46  */
  47 #include <sys/param.h>
  48 #include <sys/systm.h>
  49 #include <sys/buf.h>
  50 #include <sys/conf.h>
  51 #include <sys/device.h>
  52 #include <sys/devicestat.h>
  53 #include <sys/disk.h>
  54 #include <sys/kernel.h>
  55 #include <sys/malloc.h>
  56 #include <sys/sysctl.h>
  57 #include <sys/proc.h>
  58 #include <sys/queue.h>
  59 #include <sys/tree.h>
  60 #include <sys/udev.h>
  61 #include <sys/uuid.h>
  62 #include <sys/kern_syscall.h>
  63
  64 #include <sys/dmsg.h>
  65 #include <sys/xdiskioctl.h>
  66
  67 #include <sys/buf2.h>
  68 #include <sys/thread2.h>
  69
  70 struct xa_softc;
  71 struct xa_softc_tree;
  72 RB_HEAD(xa_softc_tree, xa_softc);
  73 RB_PROTOTYPE(xa_softc_tree, xa_softc, rbnode, xa_softc_cmp);
  74
  75 /*
  76  * Track a BIO tag
  77  */
  78 struct xa_tag {
  79         TAILQ_ENTRY(xa_tag) entry;
  80         struct xa_softc *sc;
  81         dmsg_blk_error_t status;
  82         kdmsg_state_t   *state;
  83         struct bio      *bio;
  84         int             waiting;
  85         int             async;
  86         int             done;
  87 };
  88
  89 typedef struct xa_tag   xa_tag_t;
  90
  91 /*
  92  * Track devices.
  93  */
  94 struct xa_softc {
  95         struct kdmsg_state_list spanq;
  96         RB_ENTRY(xa_softc) rbnode;
  97         cdev_t          dev;
  98         struct disk_info info;
  99         struct disk     disk;
 100         uuid_t          pfs_fsid;
 101         int             unit;
 102         int             opencnt;
 103         int             spancnt;
 104         uint64_t        keyid;
 105         int             serializing;
 106         int             last_error;
 107         char            cl_label[64];   /* from LNK_SPAN cl_label (host/dev) */
 108         char            fs_label[64];   /* from LNK_SPAN fs_label (serno str) */
 109         xa_tag_t        *open_tag;
 110         TAILQ_HEAD(, bio) bioq;         /* pending BIOs */
 111         TAILQ_HEAD(, xa_tag) tag_freeq; /* available I/O tags */
 112         TAILQ_HEAD(, xa_tag) tag_pendq; /* running I/O tags */
 113         struct lwkt_token tok;
 114 };
 115
 116 typedef struct xa_softc xa_softc_t;
 117
 118 struct xa_iocom {
 119         TAILQ_ENTRY(xa_iocom) entry;
 120         kdmsg_iocom_t   iocom;
 121         xa_softc_t      dummysc;
 122 };
 123
 124 typedef struct xa_iocom xa_iocom_t;
 125
 126 static int xa_softc_cmp(xa_softc_t *sc1, xa_softc_t *sc2);
 127 RB_GENERATE(xa_softc_tree, xa_softc, rbnode, xa_softc_cmp);
 128 static struct xa_softc_tree xa_device_tree;
 129
 130 #define MAXTAGS         64      /* no real limit */
 131
 132 static int xdisk_attach(struct xdisk_attach_ioctl *xaioc);
 133 static int xdisk_detach(struct xdisk_attach_ioctl *xaioc);
 134 static void xaio_exit(kdmsg_iocom_t *iocom);
 135 static int xaio_rcvdmsg(kdmsg_msg_t *msg);
 136
 137 static void xa_terminate_check(struct xa_softc *sc);
 138
 139 static xa_tag_t *xa_setup_cmd(xa_softc_t *sc, struct bio *bio);
 140 static void xa_start(xa_tag_t *tag, kdmsg_msg_t *msg, int async);
 141 static void xa_done(xa_tag_t *tag, int wasbio);
 142 static void xa_release(xa_tag_t *tag, int wasbio);
 143 static uint32_t xa_wait(xa_tag_t *tag);
 144 static int xa_sync_completion(kdmsg_state_t *state, kdmsg_msg_t *msg);
 145 static int xa_bio_completion(kdmsg_state_t *state, kdmsg_msg_t *msg);
 146 static void xa_restart_deferred(xa_softc_t *sc);
 147
 148 MALLOC_DEFINE(M_XDISK, "Networked disk client", "Network Disks");
 149
 150 /*
 151  * Control device, issue ioctls to create xa devices.
 152  */
 153 static d_open_t xdisk_open;
 154 static d_close_t xdisk_close;
 155 static d_ioctl_t xdisk_ioctl;
 156
 157 static struct dev_ops xdisk_ops = {
 158         { "xdisk", 0, D_MPSAFE | D_TRACKCLOSE },
 159         .d_open =       xdisk_open,
 160         .d_close =      xdisk_close,
 161         .d_ioctl =      xdisk_ioctl
 162 };
 163
 164 /*
 165  * XA disk devices
 166  */
 167 static d_open_t xa_open;
 168 static d_close_t xa_close;
 169 static d_ioctl_t xa_ioctl;
 170 static d_strategy_t xa_strategy;
 171 static d_psize_t xa_size;
 172
 173 static struct dev_ops xa_ops = {
 174         { "xa", 0, D_DISK | D_CANFREE | D_MPSAFE | D_TRACKCLOSE },
 175         .d_open =       xa_open,
 176         .d_close =      xa_close,
 177         .d_ioctl =      xa_ioctl,
 178         .d_read =       physread,
 179         .d_write =      physwrite,
 180         .d_strategy =   xa_strategy,
 181         .d_psize =      xa_size
 182 };
 183
 184 static struct lwkt_token xdisk_token = LWKT_TOKEN_INITIALIZER(xdisk_token);
 185 static int xdisk_opencount;
 186 static cdev_t xdisk_dev;
 187 static TAILQ_HEAD(, xa_iocom) xaiocomq;
 188
 189 /*
 190  * Module initialization
 191  */
 192 static int
 193 xdisk_modevent(module_t mod, int type, void *data)
 194 {
 195         switch (type) {
 196         case MOD_LOAD:
 197                 TAILQ_INIT(&xaiocomq);
 198                 RB_INIT(&xa_device_tree);
 199                 xdisk_dev = make_dev(&xdisk_ops, 0,
 200                                      UID_ROOT, GID_WHEEL, 0600, "xdisk");
 201                 break;
 202         case MOD_UNLOAD:
 203         case MOD_SHUTDOWN:
 204                 if (xdisk_opencount || TAILQ_FIRST(&xaiocomq))
 205                         return (EBUSY);
 206                 if (xdisk_dev) {
 207                         destroy_dev(xdisk_dev);
 208                         xdisk_dev = NULL;
 209                 }
 210                 dev_ops_remove_all(&xdisk_ops);
 211                 dev_ops_remove_all(&xa_ops);
 212                 break;
 213         default:
 214                 break;
 215         }
 216         return 0;
 217 }
 218
 219 DEV_MODULE(xdisk, xdisk_modevent, 0);
 220
 221 static int
 222 xa_softc_cmp(xa_softc_t *sc1, xa_softc_t *sc2)
 223 {
 224         return(bcmp(sc1->fs_label, sc2->fs_label, sizeof(sc1->fs_label)));
 225 }
 226
 227 /*
 228  * Control device
 229  */
 230 static int
 231 xdisk_open(struct dev_open_args *ap)
 232 {
 233         lwkt_gettoken(&xdisk_token);
 234         ++xdisk_opencount;
 235         lwkt_reltoken(&xdisk_token);
 236         return(0);
 237 }
 238
 239 static int
 240 xdisk_close(struct dev_close_args *ap)
 241 {
 242         lwkt_gettoken(&xdisk_token);
 243         --xdisk_opencount;
 244         lwkt_reltoken(&xdisk_token);
 245         return(0);
 246 }
 247
 248 static int
 249 xdisk_ioctl(struct dev_ioctl_args *ap)
 250 {
 251         int error;
 252
 253         switch(ap->a_cmd) {
 254         case XDISKIOCATTACH:
 255                 error = xdisk_attach((void *)ap->a_data);
 256                 break;
 257         case XDISKIOCDETACH:
 258                 error = xdisk_detach((void *)ap->a_data);
 259                 break;
 260         default:
 261                 error = ENOTTY;
 262                 break;
 263         }
 264         return error;
 265 }
 266
 267 /************************************************************************
 268  *                              DMSG INTERFACE                          *
 269  ************************************************************************/
 270
 271 static int
 272 xdisk_attach(struct xdisk_attach_ioctl *xaioc)
 273 {
 274         xa_iocom_t *xaio;
 275         struct file *fp;
 276
 277         /*
 278          * Normalize ioctl params
 279          */
 280         kprintf("xdisk_attach1\n");
 281         fp = holdfp(curproc->p_fd, xaioc->fd, -1);
 282         if (fp == NULL)
 283                 return EINVAL;
 284         kprintf("xdisk_attach2\n");
 285
 286         /*
 287          * See if the serial number is already present.  If we are
 288          * racing a termination the disk subsystem may still have
 289          * duplicate entries not yet removed so we wait a bit and
 290          * retry.
 291          */
 292         lwkt_gettoken(&xdisk_token);
 293
 294         xaio = kmalloc(sizeof(*xaio), M_XDISK, M_WAITOK | M_ZERO);
 295         kprintf("xdisk_attach3\n");
 296         kdmsg_iocom_init(&xaio->iocom, xaio,
 297                          KDMSG_IOCOMF_AUTOCONN,
 298                          M_XDISK, xaio_rcvdmsg);
 299         xaio->iocom.exit_func = xaio_exit;
 300
 301         kdmsg_iocom_reconnect(&xaio->iocom, fp, "xdisk");
 302
 303         /*
 304          * Setup our LNK_CONN advertisement for autoinitiate.
 305          *
 306          * Our filter is setup to only accept PEER_BLOCK/SERVER
 307          * advertisements.
 308          *
 309          * We need a unique pfs_fsid to avoid confusion.
 310          */
 311         xaio->iocom.auto_lnk_conn.pfs_type = DMSG_PFSTYPE_CLIENT;
 312         xaio->iocom.auto_lnk_conn.proto_version = DMSG_SPAN_PROTO_1;
 313         xaio->iocom.auto_lnk_conn.peer_type = DMSG_PEER_BLOCK;
 314         xaio->iocom.auto_lnk_conn.peer_mask = 1LLU << DMSG_PEER_BLOCK;
 315         xaio->iocom.auto_lnk_conn.pfs_mask = 1LLU << DMSG_PFSTYPE_SERVER;
 316         ksnprintf(xaio->iocom.auto_lnk_conn.fs_label,
 317                   sizeof(xaio->iocom.auto_lnk_conn.fs_label),
 318                   "xdisk");
 319         kern_uuidgen(&xaio->iocom.auto_lnk_conn.pfs_fsid, 1);
 320
 321         /*
 322          * Setup our LNK_SPAN advertisement for autoinitiate
 323          */
 324         TAILQ_INSERT_TAIL(&xaiocomq, xaio, entry);
 325         kdmsg_iocom_autoinitiate(&xaio->iocom, NULL);
 326         lwkt_reltoken(&xdisk_token);
 327
 328         return 0;
 329 }
 330
 331 static int
 332 xdisk_detach(struct xdisk_attach_ioctl *xaioc)
 333 {
 334         return EINVAL;
 335 }
 336
 337 /*
 338  * Called from iocom core transmit thread upon disconnect.
 339  */
 340 static
 341 void
 342 xaio_exit(kdmsg_iocom_t *iocom)
 343 {
 344         xa_iocom_t *xaio = iocom->handle;
 345
 346         kprintf("xdisk_detach -xaio_exit\n");
 347         lwkt_gettoken(&xdisk_token);
 348         TAILQ_REMOVE(&xaiocomq, xaio, entry);
 349         lwkt_reltoken(&xdisk_token);
 350
 351         kfree(xaio, M_XDISK);
 352 }
 353
 354 /*
 355  * Called from iocom core to handle messages that the iocom core does not
 356  * handle itself and for which a state function callback has not yet been
 357  * established.
 358  *
 359  * We primarily care about LNK_SPAN transactions here.
 360  */
 361 static int
 362 xaio_rcvdmsg(kdmsg_msg_t *msg)
 363 {
 364         kdmsg_state_t   *state = msg->state;
 365         xa_iocom_t      *xaio = state->iocom->handle;
 366         xa_softc_t      *sc;
 367
 368         kprintf("xdisk_rcvdmsg %08x\n", msg->any.head.cmd);
 369         lwkt_gettoken(&xdisk_token);
 370
 371         switch(msg->tcmd) {
 372         case DMSG_LNK_SPAN | DMSGF_CREATE | DMSGF_DELETE:
 373                 /*
 374                  * A LNK_SPAN transaction which is opened and closed
 375                  * degenerately is not useful to us, just ignore it.
 376                  */
 377                 kdmsg_msg_reply(msg, 0);
 378                 break;
 379         case DMSG_LNK_SPAN | DMSGF_CREATE:
 380                 /*
 381                  * Manage the tracking node for the remote LNK_SPAN.
 382                  *
 383                  * Return a streaming result, leaving the transaction open
 384                  * in both directions to allow sub-transactions.
 385                  */
 386                 bcopy(msg->any.lnk_span.cl_label, xaio->dummysc.cl_label,
 387                       sizeof(xaio->dummysc.cl_label));
 388                 xaio->dummysc.cl_label[sizeof(xaio->dummysc.cl_label) - 1] = 0;
 389
 390                 bcopy(msg->any.lnk_span.fs_label, xaio->dummysc.fs_label,
 391                       sizeof(xaio->dummysc.fs_label));
 392                 xaio->dummysc.fs_label[sizeof(xaio->dummysc.fs_label) - 1] = 0;
 393
 394                 kprintf("xdisk: %s LNK_SPAN create\n",
 395                         msg->any.lnk_span.fs_label);
 396
 397                 sc = RB_FIND(xa_softc_tree, &xa_device_tree, &xaio->dummysc);
 398                 if (sc == NULL) {
 399                         xa_softc_t *sctmp;
 400                         xa_tag_t *tag;
 401                         cdev_t dev;
 402                         int unit;
 403                         int n;
 404
 405                         sc = kmalloc(sizeof(*sc), M_XDISK, M_WAITOK | M_ZERO);
 406                         bcopy(msg->any.lnk_span.cl_label, sc->cl_label,
 407                               sizeof(sc->cl_label));
 408                         sc->cl_label[sizeof(sc->cl_label) - 1] = 0;
 409                         bcopy(msg->any.lnk_span.fs_label, sc->fs_label,
 410                               sizeof(sc->fs_label));
 411                         sc->fs_label[sizeof(sc->fs_label) - 1] = 0;
 412
 413                         /* XXX FIXME O(N^2) */
 414                         unit = -1;
 415                         do {
 416                                 ++unit;
 417                                 RB_FOREACH(sctmp, xa_softc_tree,
 418                                            &xa_device_tree) {
 419                                         if (sctmp->unit == unit)
 420                                                 break;
 421                                 }
 422                         } while (sctmp);
 423
 424                         sc->unit = unit;
 425                         sc->serializing = 1;
 426                         sc->spancnt = 1;
 427                         lwkt_token_init(&sc->tok, "xa");
 428                         TAILQ_INIT(&sc->spanq);
 429                         TAILQ_INIT(&sc->bioq);
 430                         TAILQ_INIT(&sc->tag_freeq);
 431                         TAILQ_INIT(&sc->tag_pendq);
 432                         RB_INSERT(xa_softc_tree, &xa_device_tree, sc);
 433                         TAILQ_INSERT_TAIL(&sc->spanq, msg->state, user_entry);
 434                         msg->state->any.xa_sc = sc;
 435
 436                         /*
 437                          * Setup block device
 438                          */
 439                         for (n = 0; n < MAXTAGS; ++n) {
 440                                 tag = kmalloc(sizeof(*tag),
 441                                               M_XDISK, M_WAITOK|M_ZERO);
 442                                 tag->sc = sc;
 443                                 TAILQ_INSERT_TAIL(&sc->tag_freeq, tag, entry);
 444                         }
 445
 446                         if (sc->dev == NULL) {
 447                                 dev = disk_create(unit, &sc->disk, &xa_ops);
 448                                 dev->si_drv1 = sc;
 449                                 sc->dev = dev;
 450                         }
 451
 452                         sc->info.d_media_blksize =
 453                                 msg->any.lnk_span.media.block.blksize;
 454                         if (sc->info.d_media_blksize <= 0)
 455                                 sc->info.d_media_blksize = 1;
 456                         sc->info.d_media_blocks =
 457                                 msg->any.lnk_span.media.block.bytes /
 458                                 sc->info.d_media_blksize;
 459                         sc->info.d_dsflags = DSO_MBRQUIET | DSO_RAWPSIZE;
 460                         sc->info.d_secpertrack = 32;
 461                         sc->info.d_nheads = 64;
 462                         sc->info.d_secpercyl = sc->info.d_secpertrack *
 463                                                sc->info.d_nheads;
 464                         sc->info.d_ncylinders = 0;
 465                         if (sc->fs_label[0])
 466                                 sc->info.d_serialno = sc->fs_label;
 467                         disk_setdiskinfo_sync(&sc->disk, &sc->info);
 468                         xa_restart_deferred(sc);        /* eats serializing */
 469                 } else {
 470                         ++sc->spancnt;
 471                         TAILQ_INSERT_TAIL(&sc->spanq, msg->state, user_entry);
 472                         msg->state->any.xa_sc = sc;
 473                         if (sc->serializing == 0 && sc->open_tag == NULL) {
 474                                 sc->serializing = 1;
 475                                 xa_restart_deferred(sc); /* eats serializing */
 476                         }
 477                 }
 478                 kdmsg_msg_result(msg, 0);
 479                 break;
 480         case DMSG_LNK_SPAN | DMSGF_DELETE:
 481         case DMSG_LNK_SPAN | DMSGF_DELETE | DMSGF_REPLY:
 482                 /*
 483                  * Manage the tracking node for the remote LNK_SPAN.
 484                  *
 485                  * Return a final result, closing our end of the transaction.
 486                  */
 487                 sc = msg->state->any.xa_sc;
 488                 kprintf("xdisk: %s LNK_SPAN terminate\n", sc->fs_label);
 489                 msg->state->any.xa_sc = NULL;
 490                 TAILQ_REMOVE(&sc->spanq, msg->state, user_entry);
 491                 --sc->spancnt;
 492                 xa_terminate_check(sc);
 493                 kdmsg_msg_reply(msg, 0);
 494                 break;
 495         case DMSG_LNK_SPAN | DMSGF_REPLY:
 496                 /*
 497                  * Ignore unimplemented streaming replies on our LNK_SPAN
 498                  * transaction.
 499                  */
 500                 break;
 501         case DMSG_DBG_SHELL:
 502                 /*
 503                  * Execute shell command (not supported atm).
 504                  *
 505                  * This is a one-way packet but if not (e.g. if part of
 506                  * a streaming transaction), we will have already closed
 507                  * our end.
 508                  */
 509                 kdmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
 510                 break;
 511         case DMSG_DBG_SHELL | DMSGF_REPLY:
 512                 /*
 513                  * Receive one or more replies to a shell command
 514                  * that we sent.  Just dump it to the console.
 515                  *
 516                  * This is a one-way packet but if not (e.g. if
 517                  * part of a streaming transaction), we will have
 518                  * already closed our end.
 519                  */
 520                 if (msg->aux_data) {
 521                         msg->aux_data[msg->aux_size - 1] = 0;
 522                         kprintf("xdisk: DEBUGMSG: %s\n",
 523                                 msg->aux_data);
 524                 }
 525                 break;
 526         default:
 527                 /*
 528                  * Unsupported one-way message, streaming message, or
 529                  * transaction.
 530                  *
 531                  * Terminate any unsupported transactions with an error
 532                  * and ignore any unsupported streaming messages.
 533                  *
 534                  * NOTE: This case also includes DMSG_LNK_ERROR messages
 535                  *       which might be one-way, replying to those would
 536                  *       cause an infinite ping-pong.
 537                  */
 538                 if (msg->any.head.cmd & DMSGF_CREATE)
 539                         kdmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
 540                 break;
 541         }
 542         lwkt_reltoken(&xdisk_token);
 543
 544         return 0;
 545 }
 546
 547 /*
 548  * Determine if we can destroy the xa_softc.
 549  *
 550  * Called with xdisk_token held.
 551  */
 552 static
 553 void
 554 xa_terminate_check(struct xa_softc *sc)
 555 {
 556         xa_tag_t *tag;
 557
 558         /*
 559          * Determine if we can destroy the softc.
 560          */
 561         kprintf("xdisk: terminate check xa%d (%d,%d,%d)\n",
 562                 sc->unit,
 563                 sc->opencnt, sc->serializing, sc->spancnt);
 564
 565         if (sc->opencnt || sc->serializing || sc->spancnt)
 566                 return;
 567         sc->serializing = 1;
 568         KKASSERT(TAILQ_EMPTY(&sc->tag_pendq));
 569
 570         RB_REMOVE(xa_softc_tree, &xa_device_tree, sc);
 571
 572         if (sc->dev) {
 573                 disk_destroy(&sc->disk);
 574                 sc->dev->si_drv1 = NULL;
 575                 sc->dev = NULL;
 576         }
 577         KKASSERT(sc->opencnt == 0);
 578         KKASSERT(TAILQ_EMPTY(&sc->tag_pendq));
 579
 580         while ((tag = TAILQ_FIRST(&sc->tag_freeq)) != NULL) {
 581                 TAILQ_REMOVE(&sc->tag_freeq, tag, entry);
 582                 tag->sc = NULL;
 583                 kfree(tag, M_XDISK);
 584         }
 585         kfree(sc, M_XDISK);
 586 }
 587
 588 /************************************************************************
 589  *                         XA DEVICE INTERFACE                          *
 590  ************************************************************************/
 591
 592 static int
 593 xa_open(struct dev_open_args *ap)
 594 {
 595         cdev_t dev = ap->a_head.a_dev;
 596         xa_softc_t *sc;
 597         int error;
 598
 599         dev->si_bsize_phys = 512;
 600         dev->si_bsize_best = 32768;
 601
 602         /*
 603          * Interlock open with opencnt, wait for attachment operations
 604          * to finish.
 605          */
 606         lwkt_gettoken(&xdisk_token);
 607 again:
 608         sc = dev->si_drv1;
 609         if (sc == NULL) {
 610                 lwkt_reltoken(&xdisk_token);
 611                 return ENXIO;   /* raced destruction */
 612         }
 613         if (sc->serializing) {
 614                 tsleep(sc, 0, "xarace", hz / 10);
 615                 goto again;
 616         }
 617         sc->serializing = 1;
 618
 619         /*
 620          * Serialize initial open
 621          */
 622         if (sc->opencnt++ > 0) {
 623                 lwkt_reltoken(&xdisk_token);
 624                 return(0);
 625         }
 626         lwkt_reltoken(&xdisk_token);
 627
 628         /*
 629          * Issue BLK_OPEN if necessary.  ENXIO is returned if we have trouble.
 630          */
 631         if (sc->open_tag == NULL) {
 632                 xa_restart_deferred(sc); /* eats serializing */
 633         } else {
 634                 sc->serializing = 0;
 635                 wakeup(sc);
 636         }
 637
 638         /*
 639          * Wait for completion of the BLK_OPEN
 640          */
 641         lwkt_gettoken(&xdisk_token);
 642         while (sc->serializing)
 643                 tsleep(sc, 0, "xaopen", hz);
 644
 645         error = sc->last_error;
 646         if (error) {
 647                 KKASSERT(sc->opencnt > 0);
 648                 --sc->opencnt;
 649                 xa_terminate_check(sc);
 650                 sc = NULL;      /* sc may be invalid now */
 651         }
 652         lwkt_reltoken(&xdisk_token);
 653
 654         return (error);
 655 }
 656
 657 static int
 658 xa_close(struct dev_close_args *ap)
 659 {
 660         cdev_t dev = ap->a_head.a_dev;
 661         xa_softc_t *sc;
 662         xa_tag_t *tag;
 663
 664         sc = dev->si_drv1;
 665         if (sc == NULL)
 666                 return ENXIO;   /* raced destruction */
 667         lwkt_gettoken(&xdisk_token);
 668         lwkt_gettoken(&sc->tok);
 669
 670         /*
 671          * NOTE: Clearing open_tag allows a concurrent open to re-open
 672          *       the device and prevents autonomous completion of the tag.
 673          */
 674         if (sc->opencnt == 1 && sc->open_tag) {
 675                 tag = sc->open_tag;
 676                 sc->open_tag = NULL;
 677                 kdmsg_state_reply(tag->state, 0);       /* close our side */
 678                 xa_wait(tag);                           /* wait on remote */
 679         }
 680         lwkt_reltoken(&sc->tok);
 681         KKASSERT(sc->opencnt > 0);
 682         --sc->opencnt;
 683         xa_terminate_check(sc);
 684         lwkt_reltoken(&xdisk_token);
 685
 686         return(0);
 687 }
 688
 689 static int
 690 xa_strategy(struct dev_strategy_args *ap)
 691 {
 692         xa_softc_t *sc = ap->a_head.a_dev->si_drv1;
 693         xa_tag_t *tag;
 694         struct bio *bio = ap->a_bio;
 695
 696         /*
 697          * Allow potentially temporary link failures to fail the I/Os
 698          * only if the device is not open.  That is, we allow the disk
 699          * probe code prior to mount to fail.
 700          */
 701         if (sc->opencnt == 0) {
 702                 bio->bio_buf->b_error = ENXIO;
 703                 bio->bio_buf->b_flags |= B_ERROR;
 704                 biodone(bio);
 705                 return(0);
 706         }
 707
 708         tag = xa_setup_cmd(sc, bio);
 709         if (tag)
 710                 xa_start(tag, NULL, 1);
 711         return(0);
 712 }
 713
 714 static int
 715 xa_ioctl(struct dev_ioctl_args *ap)
 716 {
 717         return(ENOTTY);
 718 }
 719
 720 static int
 721 xa_size(struct dev_psize_args *ap)
 722 {
 723         struct xa_softc *sc;
 724
 725         if ((sc = ap->a_head.a_dev->si_drv1) == NULL)
 726                 return (ENXIO);
 727         ap->a_result = sc->info.d_media_blocks;
 728         return (0);
 729 }
 730
 731 /************************************************************************
 732  *                  XA BLOCK PROTOCOL STATE MACHINE                     *
 733  ************************************************************************
 734  *
 735  * Implement tag/msg setup and related functions.
 736  */
 737 static xa_tag_t *
 738 xa_setup_cmd(xa_softc_t *sc, struct bio *bio)
 739 {
 740         xa_tag_t *tag;
 741
 742         /*
 743          * Only get a tag if we have a valid virtual circuit to the server.
 744          */
 745         lwkt_gettoken(&sc->tok);
 746         if ((tag = TAILQ_FIRST(&sc->tag_freeq)) != NULL) {
 747                 TAILQ_REMOVE(&sc->tag_freeq, tag, entry);
 748                 tag->bio = bio;
 749                 TAILQ_INSERT_TAIL(&sc->tag_pendq, tag, entry);
 750         }
 751
 752         /*
 753          * If we can't dispatch now and this is a bio, queue it for later.
 754          */
 755         if (tag == NULL && bio) {
 756                 TAILQ_INSERT_TAIL(&sc->bioq, bio, bio_act);
 757         }
 758         lwkt_reltoken(&sc->tok);
 759
 760         return (tag);
 761 }
 762
 763 static void
 764 xa_start(xa_tag_t *tag, kdmsg_msg_t *msg, int async)
 765 {
 766         xa_softc_t *sc = tag->sc;
 767
 768         tag->done = 0;
 769         tag->async = async;
 770
 771         if (msg == NULL) {
 772                 struct bio *bio;
 773                 struct buf *bp;
 774
 775                 KKASSERT(tag->bio);
 776                 bio = tag->bio;
 777                 bp = bio->bio_buf;
 778
 779                 switch(bp->b_cmd) {
 780                 case BUF_CMD_READ:
 781                         msg = kdmsg_msg_alloc(sc->open_tag->state,
 782                                               DMSG_BLK_READ |
 783                                               DMSGF_CREATE | DMSGF_DELETE,
 784                                               xa_bio_completion, tag);
 785                         msg->any.blk_read.keyid = sc->keyid;
 786                         msg->any.blk_read.offset = bio->bio_offset;
 787                         msg->any.blk_read.bytes = bp->b_bcount;
 788                         break;
 789                 case BUF_CMD_WRITE:
 790                         msg = kdmsg_msg_alloc(sc->open_tag->state,
 791                                               DMSG_BLK_WRITE |
 792                                               DMSGF_CREATE | DMSGF_DELETE,
 793                                               xa_bio_completion, tag);
 794                         msg->any.blk_write.keyid = sc->keyid;
 795                         msg->any.blk_write.offset = bio->bio_offset;
 796                         msg->any.blk_write.bytes = bp->b_bcount;
 797                         msg->aux_data = bp->b_data;
 798                         msg->aux_size = bp->b_bcount;
 799                         break;
 800                 case BUF_CMD_FLUSH:
 801                         msg = kdmsg_msg_alloc(sc->open_tag->state,
 802                                               DMSG_BLK_FLUSH |
 803                                               DMSGF_CREATE | DMSGF_DELETE,
 804                                               xa_bio_completion, tag);
 805                         msg->any.blk_flush.keyid = sc->keyid;
 806                         msg->any.blk_flush.offset = bio->bio_offset;
 807                         msg->any.blk_flush.bytes = bp->b_bcount;
 808                         break;
 809                 case BUF_CMD_FREEBLKS:
 810                         msg = kdmsg_msg_alloc(sc->open_tag->state,
 811                                               DMSG_BLK_FREEBLKS |
 812                                               DMSGF_CREATE | DMSGF_DELETE,
 813                                               xa_bio_completion, tag);
 814                         msg->any.blk_freeblks.keyid = sc->keyid;
 815                         msg->any.blk_freeblks.offset = bio->bio_offset;
 816                         msg->any.blk_freeblks.bytes = bp->b_bcount;
 817                         break;
 818                 default:
 819                         bp->b_flags |= B_ERROR;
 820                         bp->b_error = EIO;
 821                         biodone(bio);
 822                         tag->bio = NULL;
 823                         break;
 824                 }
 825         }
 826
 827         if (msg) {
 828                 tag->state = msg->state;
 829                 kdmsg_msg_write(msg);
 830         } else {
 831                 tag->status.head.error = DMSG_ERR_IO;
 832                 xa_done(tag, 1);
 833         }
 834 }
 835
 836 static uint32_t
 837 xa_wait(xa_tag_t *tag)
 838 {
 839         xa_softc_t *sc = tag->sc;
 840         uint32_t error;
 841
 842         kprintf("xdisk: xa_wait  %p\n", tag);
 843
 844         lwkt_gettoken(&sc->tok);
 845         tag->waiting = 1;
 846         while (tag->done == 0)
 847                 tsleep(tag, 0, "xawait", 0);
 848         lwkt_reltoken(&sc->tok);
 849         error = tag->status.head.error;
 850         tag->waiting = 0;
 851         xa_release(tag, 0);
 852
 853         return error;
 854 }
 855
 856 static void
 857 xa_done(xa_tag_t *tag, int wasbio)
 858 {
 859         KKASSERT(tag->bio == NULL);
 860
 861         tag->state = NULL;
 862         tag->done = 1;
 863         if (tag->waiting)
 864                 wakeup(tag);
 865         if (tag->async)
 866                 xa_release(tag, wasbio);
 867 }
 868
 869 static
 870 void
 871 xa_release(xa_tag_t *tag, int wasbio)
 872 {
 873         xa_softc_t *sc = tag->sc;
 874         struct bio *bio;
 875
 876         lwkt_gettoken(&sc->tok);
 877         if (wasbio && (bio = TAILQ_FIRST(&sc->bioq)) != NULL) {
 878                 TAILQ_REMOVE(&sc->bioq, bio, bio_act);
 879                 tag->bio = bio;
 880                 lwkt_reltoken(&sc->tok);
 881                 xa_start(tag, NULL, 1);
 882         } else {
 883                 TAILQ_REMOVE(&sc->tag_pendq, tag, entry);
 884                 TAILQ_INSERT_TAIL(&sc->tag_freeq, tag, entry);
 885                 lwkt_reltoken(&sc->tok);
 886         }
 887 }
 888
 889 /*
 890  * Handle messages under the BLKOPEN transaction.
 891  */
 892 static int
 893 xa_sync_completion(kdmsg_state_t *state, kdmsg_msg_t *msg)
 894 {
 895         xa_tag_t *tag = state->any.any;
 896         xa_softc_t *sc = tag->sc;
 897         struct bio *bio;
 898
 899         /*
 900          * If the tag has been cleaned out we already closed our side
 901          * of the transaction and we are waiting for the other side to
 902          * close.
 903          */
 904         if (tag == NULL) {
 905                 if (msg->any.head.cmd & DMSGF_CREATE)
 906                         kdmsg_state_reply(state, DMSG_ERR_LOSTLINK);
 907                 return 0;
 908         }
 909
 910         /*
 911          * Validate the tag
 912          */
 913         lwkt_gettoken(&sc->tok);
 914
 915         /*
 916          * Handle initial response to our open and restart any deferred
 917          * BIOs on success.
 918          *
 919          * NOTE: DELETE may also be set.
 920          */
 921         if (msg->any.head.cmd & DMSGF_CREATE) {
 922                 switch(msg->any.head.cmd & DMSGF_CMDSWMASK) {
 923                 case DMSG_LNK_ERROR | DMSGF_REPLY:
 924                         bzero(&tag->status, sizeof(tag->status));
 925                         tag->status.head = msg->any.head;
 926                         break;
 927                 case DMSG_BLK_ERROR | DMSGF_REPLY:
 928                         tag->status = msg->any.blk_error;
 929                         break;
 930                 }
 931                 sc->last_error = tag->status.head.error;
 932                 kprintf("xdisk: blk_open completion status %d\n",
 933                         sc->last_error);
 934                 if (sc->last_error == 0) {
 935                         while ((bio = TAILQ_FIRST(&sc->bioq)) != NULL) {
 936                                 tag = xa_setup_cmd(sc, NULL);
 937                                 if (tag == NULL)
 938                                         break;
 939                                 TAILQ_REMOVE(&sc->bioq, bio, bio_act);
 940                                 tag->bio = bio;
 941                                 xa_start(tag, NULL, 1);
 942                         }
 943                 }
 944                 sc->serializing = 0;
 945                 wakeup(sc);
 946         }
 947
 948         /*
 949          * Handle unexpected termination (or lost comm channel) from other
 950          * side.  Autonomous completion only if open_tag matches,
 951          * otherwise another thread is probably waiting on the tag.
 952          *
 953          * (see xa_close() for other interactions)
 954          */
 955         if (msg->any.head.cmd & DMSGF_DELETE) {
 956                 kdmsg_state_reply(tag->state, 0);
 957                 if (sc->open_tag == tag) {
 958                         sc->open_tag = NULL;
 959                         xa_done(tag, 0);
 960                 } else {
 961                         tag->async = 0;
 962                         xa_done(tag, 0);
 963                 }
 964         }
 965         lwkt_reltoken(&sc->tok);
 966         return (0);
 967 }
 968
 969 static int
 970 xa_bio_completion(kdmsg_state_t *state, kdmsg_msg_t *msg)
 971 {
 972         xa_tag_t *tag = state->any.any;
 973         xa_softc_t *sc = tag->sc;
 974         struct bio *bio;
 975         struct buf *bp;
 976
 977         /*
 978          * Get the bio from the tag.  If no bio is present we just do
 979          * 'done' handling.
 980          */
 981         if ((bio = tag->bio) == NULL)
 982                 goto handle_done;
 983         bp = bio->bio_buf;
 984
 985         /*
 986          * Process return status
 987          */
 988         switch(msg->any.head.cmd & DMSGF_CMDSWMASK) {
 989         case DMSG_LNK_ERROR | DMSGF_REPLY:
 990                 bzero(&tag->status, sizeof(tag->status));
 991                 tag->status.head = msg->any.head;
 992                 if (tag->status.head.error)
 993                         tag->status.resid = bp->b_bcount;
 994                 else
 995                         tag->status.resid = 0;
 996                 break;
 997         case DMSG_BLK_ERROR | DMSGF_REPLY:
 998                 tag->status = msg->any.blk_error;
 999                 break;
1000         }
1001
1002         /*
1003          * Potentially move the bio back onto the pending queue if the
1004          * device is open and the error is related to losing the virtual
1005          * circuit.
1006          */
1007         if (tag->status.head.error &&
1008             (msg->any.head.cmd & DMSGF_DELETE) && sc->opencnt) {
1009                 if (tag->status.head.error == DMSG_ERR_LOSTLINK ||
1010                     tag->status.head.error == DMSG_ERR_CANTCIRC) {
1011                         goto handle_repend;
1012                 }
1013         }
1014
1015         /*
1016          * Process bio completion
1017          *
1018          * For reads any returned data is zero-extended if necessary, so
1019          * the server can short-cut any all-zeros reads if it desires.
1020          */
1021         switch(bp->b_cmd) {
1022         case BUF_CMD_READ:
1023                 if (msg->aux_data && msg->aux_size) {
1024                         if (msg->aux_size < bp->b_bcount) {
1025                                 bcopy(msg->aux_data, bp->b_data, msg->aux_size);
1026                                 bzero(bp->b_data + msg->aux_size,
1027                                       bp->b_bcount - msg->aux_size);
1028                         } else {
1029                                 bcopy(msg->aux_data, bp->b_data, bp->b_bcount);
1030                         }
1031                 } else {
1032                         bzero(bp->b_data, bp->b_bcount);
1033                 }
1034                 /* fall through */
1035         case BUF_CMD_WRITE:
1036         case BUF_CMD_FLUSH:
1037         case BUF_CMD_FREEBLKS:
1038         default:
1039                 if (tag->status.resid > bp->b_bcount)
1040                         tag->status.resid = bp->b_bcount;
1041                 bp->b_resid = tag->status.resid;
1042                 if ((bp->b_error = tag->status.head.error) != 0) {
1043                         bp->b_flags |= B_ERROR;
1044                 } else {
1045                         bp->b_resid = 0;
1046                 }
1047                 biodone(bio);
1048                 tag->bio = NULL;
1049                 break;
1050         }
1051
1052         /*
1053          * Handle completion of the transaction.  If the bioq is not empty
1054          * we can initiate another bio on the same tag.
1055          *
1056          * NOTE: Most of our transactions will be single-message
1057          *       CREATE+DELETEs, so we won't have to terminate the
1058          *       transaction separately, here.  But just in case they
1059          *       aren't be sure to terminate the transaction.
1060          */
1061 handle_done:
1062         if (msg->any.head.cmd & DMSGF_DELETE) {
1063                 xa_done(tag, 1);
1064                 if ((state->txcmd & DMSGF_DELETE) == 0)
1065                         kdmsg_msg_reply(msg, 0);
1066         }
1067         return (0);
1068
1069         /*
1070          * Handle the case where the transaction failed due to a
1071          * connectivity issue.  The tag is put away with wasbio=0
1072          * and we put the BIO back onto the bioq for a later restart.
1073          */
1074 handle_repend:
1075         lwkt_gettoken(&sc->tok);
1076         kprintf("BIO CIRC FAILURE, REPEND BIO %p\n", bio);
1077         tag->bio = NULL;
1078         xa_done(tag, 0);
1079         if ((state->txcmd & DMSGF_DELETE) == 0)
1080                 kdmsg_msg_reply(msg, 0);
1081
1082         /*
1083          * Requeue the bio
1084          */
1085         TAILQ_INSERT_TAIL(&sc->bioq, bio, bio_act);
1086
1087         lwkt_reltoken(&sc->tok);
1088         return (0);
1089 }
1090
1091 /*
1092  * Restart as much deferred I/O as we can.  The serializer is set and we
1093  * eat it (clear it) when done.
1094  *
1095  * Called with sc->tok held
1096  */
1097 static
1098 void
1099 xa_restart_deferred(xa_softc_t *sc)
1100 {
1101         kdmsg_state_t *span;
1102         kdmsg_msg_t *msg;
1103         xa_tag_t *tag;
1104         int error;
1105
1106         KKASSERT(sc->serializing);
1107
1108         /*
1109          * Determine if a restart is needed.
1110          */
1111         if (sc->opencnt == 0) {
1112                 /*
1113                  * Device is not open, nothing to do, eat serializing.
1114                  */
1115                 sc->serializing = 0;
1116                 wakeup(sc);
1117         } else if (sc->open_tag == NULL) {
1118                 /*
1119                  * BLK_OPEN required before we can restart any BIOs.
1120                  * Select the best LNK_SPAN to issue the BLK_OPEN under.
1121                  *
1122                  * serializing interlocks waiting open()s.
1123                  */
1124                 error = 0;
1125                 TAILQ_FOREACH(span, &sc->spanq, user_entry) {
1126                         if ((span->rxcmd & DMSGF_DELETE) == 0)
1127                                 break;
1128                 }
1129                 if (span == NULL)
1130                         error = ENXIO;
1131
1132                 if (error == 0) {
1133                         tag = xa_setup_cmd(sc, NULL);
1134                         if (tag == NULL)
1135                                 error = ENXIO;
1136                 }
1137                 if (error == 0) {
1138                         kprintf("xdisk: BLK_OPEN\n");
1139                         sc->open_tag = tag;
1140                         msg = kdmsg_msg_alloc(span,
1141                                               DMSG_BLK_OPEN |
1142                                               DMSGF_CREATE,
1143                                               xa_sync_completion, tag);
1144                         msg->any.blk_open.modes = DMSG_BLKOPEN_RD;
1145                         xa_start(tag, msg, 0);
1146                 }
1147                 if (error) {
1148                         sc->serializing = 0;
1149                         wakeup(sc);
1150                 }
1151                 /* else leave serializing set until BLK_OPEN response */
1152         } else {
1153                 /* nothing to do */
1154                 sc->serializing = 0;
1155                 wakeup(sc);
1156         }
1157 }