xdisk - Fix improper EIO
[dragonfly.git] / 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 static int xa_active;
76 SYSCTL_INT(_debug, OID_AUTO, xa_active, CTLFLAG_RW, &xa_active, 0,
77            "Number of active xdisk IOs");
78 static uint64_t xa_last;
79 SYSCTL_ULONG(_debug, OID_AUTO, xa_last, CTLFLAG_RW, &xa_last, 0,
80            "Offset of last xdisk IO");
81
82 /*
83  * Track a BIO tag
84  */
85 struct xa_tag {
86         TAILQ_ENTRY(xa_tag) entry;
87         struct xa_softc *sc;
88         dmsg_blk_error_t status;
89         kdmsg_state_t   *state;
90         struct bio      *bio;
91         int             waiting;
92         int             async;
93         int             done;
94 };
95
96 typedef struct xa_tag   xa_tag_t;
97
98 /*
99  * Track devices.
100  */
101 struct xa_softc {
102         struct kdmsg_state_list spanq;
103         RB_ENTRY(xa_softc) rbnode;
104         cdev_t          dev;
105         struct devstat  stats;
106         struct disk_info info;
107         struct disk     disk;
108         uuid_t          pfs_fsid;
109         int             unit;
110         int             opencnt;
111         int             spancnt;
112         uint64_t        keyid;
113         int             serializing;
114         int             last_error;
115         char            cl_label[64];   /* from LNK_SPAN cl_label (host/dev) */
116         char            fs_label[64];   /* from LNK_SPAN fs_label (serno str) */
117         xa_tag_t        *open_tag;
118         TAILQ_HEAD(, bio) bioq;         /* pending BIOs */
119         TAILQ_HEAD(, xa_tag) tag_freeq; /* available I/O tags */
120         TAILQ_HEAD(, xa_tag) tag_pendq; /* running I/O tags */
121         struct lock     lk;
122 };
123
124 typedef struct xa_softc xa_softc_t;
125
126 struct xa_iocom {
127         TAILQ_ENTRY(xa_iocom) entry;
128         kdmsg_iocom_t   iocom;
129         xa_softc_t      dummysc;
130 };
131
132 typedef struct xa_iocom xa_iocom_t;
133
134 static int xa_softc_cmp(xa_softc_t *sc1, xa_softc_t *sc2);
135 RB_GENERATE(xa_softc_tree, xa_softc, rbnode, xa_softc_cmp);
136 static struct xa_softc_tree xa_device_tree;
137
138 #define MAXTAGS         64      /* no real limit */
139
140 static int xdisk_attach(struct xdisk_attach_ioctl *xaioc);
141 static int xdisk_detach(struct xdisk_attach_ioctl *xaioc);
142 static void xaio_exit(kdmsg_iocom_t *iocom);
143 static int xaio_rcvdmsg(kdmsg_msg_t *msg);
144
145 static void xa_terminate_check(struct xa_softc *sc);
146
147 static xa_tag_t *xa_setup_cmd(xa_softc_t *sc, struct bio *bio);
148 static void xa_start(xa_tag_t *tag, kdmsg_msg_t *msg, int async);
149 static void xa_done(xa_tag_t *tag, int wasbio);
150 static void xa_release(xa_tag_t *tag, int wasbio);
151 static uint32_t xa_wait(xa_tag_t *tag);
152 static int xa_sync_completion(kdmsg_state_t *state, kdmsg_msg_t *msg);
153 static int xa_bio_completion(kdmsg_state_t *state, kdmsg_msg_t *msg);
154 static void xa_restart_deferred(xa_softc_t *sc);
155
156 MALLOC_DEFINE(M_XDISK, "Networked disk client", "Network Disks");
157
158 /*
159  * Control device, issue ioctls to create xa devices.
160  */
161 static d_open_t xdisk_open;
162 static d_close_t xdisk_close;
163 static d_ioctl_t xdisk_ioctl;
164
165 static struct dev_ops xdisk_ops = {
166         { "xdisk", 0, D_MPSAFE | D_TRACKCLOSE },
167         .d_open =       xdisk_open,
168         .d_close =      xdisk_close,
169         .d_ioctl =      xdisk_ioctl
170 };
171
172 /*
173  * XA disk devices
174  */
175 static d_open_t xa_open;
176 static d_close_t xa_close;
177 static d_ioctl_t xa_ioctl;
178 static d_strategy_t xa_strategy;
179 static d_psize_t xa_size;
180
181 static struct dev_ops xa_ops = {
182         { "xa", 0, D_DISK | D_CANFREE | D_MPSAFE | D_TRACKCLOSE },
183         .d_open =       xa_open,
184         .d_close =      xa_close,
185         .d_ioctl =      xa_ioctl,
186         .d_read =       physread,
187         .d_write =      physwrite,
188         .d_strategy =   xa_strategy,
189         .d_psize =      xa_size
190 };
191
192 static int xdisk_opencount;
193 static cdev_t xdisk_dev;
194 struct lock xdisk_lk;
195 static TAILQ_HEAD(, xa_iocom) xaiocomq;
196
197 /*
198  * Module initialization
199  */
200 static int
201 xdisk_modevent(module_t mod, int type, void *data)
202 {
203         switch (type) {
204         case MOD_LOAD:
205                 TAILQ_INIT(&xaiocomq);
206                 RB_INIT(&xa_device_tree);
207                 lockinit(&xdisk_lk, "xdisk", 0, 0);
208                 xdisk_dev = make_dev(&xdisk_ops, 0,
209                                      UID_ROOT, GID_WHEEL, 0600, "xdisk");
210                 break;
211         case MOD_UNLOAD:
212         case MOD_SHUTDOWN:
213                 if (xdisk_opencount || TAILQ_FIRST(&xaiocomq))
214                         return (EBUSY);
215                 if (xdisk_dev) {
216                         destroy_dev(xdisk_dev);
217                         xdisk_dev = NULL;
218                 }
219                 dev_ops_remove_all(&xdisk_ops);
220                 dev_ops_remove_all(&xa_ops);
221                 break;
222         default:
223                 break;
224         }
225         return 0;
226 }
227
228 DEV_MODULE(xdisk, xdisk_modevent, 0);
229
230 static int
231 xa_softc_cmp(xa_softc_t *sc1, xa_softc_t *sc2)
232 {
233         return(strcmp(sc1->fs_label, sc2->fs_label));
234 }
235
236 /*
237  * Control device
238  */
239 static int
240 xdisk_open(struct dev_open_args *ap)
241 {
242         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
243         ++xdisk_opencount;
244         lockmgr(&xdisk_lk, LK_RELEASE);
245         return(0);
246 }
247
248 static int
249 xdisk_close(struct dev_close_args *ap)
250 {
251         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
252         --xdisk_opencount;
253         lockmgr(&xdisk_lk, LK_RELEASE);
254         return(0);
255 }
256
257 static int
258 xdisk_ioctl(struct dev_ioctl_args *ap)
259 {
260         int error;
261
262         switch(ap->a_cmd) {
263         case XDISKIOCATTACH:
264                 error = xdisk_attach((void *)ap->a_data);
265                 break;
266         case XDISKIOCDETACH:
267                 error = xdisk_detach((void *)ap->a_data);
268                 break;
269         default:
270                 error = ENOTTY;
271                 break;
272         }
273         return error;
274 }
275
276 /************************************************************************
277  *                              DMSG INTERFACE                          *
278  ************************************************************************/
279
280 static int
281 xdisk_attach(struct xdisk_attach_ioctl *xaioc)
282 {
283         xa_iocom_t *xaio;
284         struct file *fp;
285
286         /*
287          * Normalize ioctl params
288          */
289         fp = holdfp(curproc->p_fd, xaioc->fd, -1);
290         if (fp == NULL)
291                 return EINVAL;
292         kprintf("xdisk_attach fp=%p\n", fp);
293
294         /*
295          * See if the serial number is already present.  If we are
296          * racing a termination the disk subsystem may still have
297          * duplicate entries not yet removed so we wait a bit and
298          * retry.
299          */
300         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
301
302         xaio = kmalloc(sizeof(*xaio), M_XDISK, M_WAITOK | M_ZERO);
303         kdmsg_iocom_init(&xaio->iocom, xaio,
304                          KDMSG_IOCOMF_AUTOCONN,
305                          M_XDISK, xaio_rcvdmsg);
306         xaio->iocom.exit_func = xaio_exit;
307
308         kdmsg_iocom_reconnect(&xaio->iocom, fp, "xdisk");
309
310         /*
311          * Setup our LNK_CONN advertisement for autoinitiate.
312          *
313          * Our filter is setup to only accept PEER_BLOCK/SERVER
314          * advertisements.
315          *
316          * We need a unique pfs_fsid to avoid confusion.
317          */
318         xaio->iocom.auto_lnk_conn.pfs_type = DMSG_PFSTYPE_CLIENT;
319         xaio->iocom.auto_lnk_conn.proto_version = DMSG_SPAN_PROTO_1;
320         xaio->iocom.auto_lnk_conn.peer_type = DMSG_PEER_BLOCK;
321         xaio->iocom.auto_lnk_conn.peer_mask = 1LLU << DMSG_PEER_BLOCK;
322         xaio->iocom.auto_lnk_conn.pfs_mask = 1LLU << DMSG_PFSTYPE_SERVER;
323         ksnprintf(xaio->iocom.auto_lnk_conn.fs_label,
324                   sizeof(xaio->iocom.auto_lnk_conn.fs_label),
325                   "xdisk");
326         kern_uuidgen(&xaio->iocom.auto_lnk_conn.pfs_fsid, 1);
327
328         /*
329          * Setup our LNK_SPAN advertisement for autoinitiate
330          */
331         TAILQ_INSERT_TAIL(&xaiocomq, xaio, entry);
332         kdmsg_iocom_autoinitiate(&xaio->iocom, NULL);
333
334         lockmgr(&xdisk_lk, LK_RELEASE);
335
336         return 0;
337 }
338
339 static int
340 xdisk_detach(struct xdisk_attach_ioctl *xaioc)
341 {
342         return EINVAL;
343 }
344
345 /*
346  * Called from iocom core transmit thread upon disconnect.
347  */
348 static
349 void
350 xaio_exit(kdmsg_iocom_t *iocom)
351 {
352         xa_iocom_t *xaio = iocom->handle;
353
354         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
355         kprintf("xdisk_detach [xaio_exit()]\n");
356         TAILQ_REMOVE(&xaiocomq, xaio, entry);
357         lockmgr(&xdisk_lk, LK_RELEASE);
358
359         kdmsg_iocom_uninit(&xaio->iocom);
360
361         kfree(xaio, M_XDISK);
362 }
363
364 /*
365  * Called from iocom core to handle messages that the iocom core does not
366  * handle itself and for which a state function callback has not yet been
367  * established.
368  *
369  * We primarily care about LNK_SPAN transactions here.
370  */
371 static int
372 xaio_rcvdmsg(kdmsg_msg_t *msg)
373 {
374         kdmsg_state_t   *state = msg->state;
375         xa_iocom_t      *xaio = state->iocom->handle;
376         xa_softc_t      *sc;
377
378         if (state) {
379                 kprintf("xdisk - rcvmsg state=%p rx=%08x tx=%08x msgcmd=%08x\n",
380                         state, state->rxcmd, state->txcmd,
381                         msg->any.head.cmd);
382         }
383         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
384
385         switch(msg->tcmd) {
386         case DMSG_LNK_SPAN | DMSGF_CREATE | DMSGF_DELETE:
387                 /*
388                  * A LNK_SPAN transaction which is opened and closed
389                  * degenerately is not useful to us, just ignore it.
390                  */
391                 kdmsg_msg_reply(msg, 0);
392                 break;
393         case DMSG_LNK_SPAN | DMSGF_CREATE:
394                 /*
395                  * Manage the tracking node for the remote LNK_SPAN.
396                  *
397                  * Return a streaming result, leaving the transaction open
398                  * in both directions to allow sub-transactions.
399                  */
400                 bcopy(msg->any.lnk_span.cl_label, xaio->dummysc.cl_label,
401                       sizeof(xaio->dummysc.cl_label));
402                 xaio->dummysc.cl_label[sizeof(xaio->dummysc.cl_label) - 1] = 0;
403
404                 bcopy(msg->any.lnk_span.fs_label, xaio->dummysc.fs_label,
405                       sizeof(xaio->dummysc.fs_label));
406                 xaio->dummysc.fs_label[sizeof(xaio->dummysc.fs_label) - 1] = 0;
407
408                 kprintf("xdisk: LINK_SPAN state %p create for %s\n",
409                         msg->state, msg->any.lnk_span.fs_label);
410
411                 sc = RB_FIND(xa_softc_tree, &xa_device_tree, &xaio->dummysc);
412                 if (sc == NULL) {
413                         xa_softc_t *sctmp;
414                         xa_tag_t *tag;
415                         cdev_t dev;
416                         int unit;
417                         int n;
418
419                         sc = kmalloc(sizeof(*sc), M_XDISK, M_WAITOK | M_ZERO);
420                         bcopy(msg->any.lnk_span.cl_label, sc->cl_label,
421                               sizeof(sc->cl_label));
422                         sc->cl_label[sizeof(sc->cl_label) - 1] = 0;
423                         bcopy(msg->any.lnk_span.fs_label, sc->fs_label,
424                               sizeof(sc->fs_label));
425                         sc->fs_label[sizeof(sc->fs_label) - 1] = 0;
426
427                         /* XXX FIXME O(N^2) */
428                         unit = -1;
429                         do {
430                                 ++unit;
431                                 RB_FOREACH(sctmp, xa_softc_tree,
432                                            &xa_device_tree) {
433                                         if (sctmp->unit == unit)
434                                                 break;
435                                 }
436                         } while (sctmp);
437
438                         sc->unit = unit;
439                         sc->serializing = 1;
440                         sc->spancnt = 1;
441                         lockinit(&sc->lk, "xalk", 0, 0);
442                         TAILQ_INIT(&sc->spanq);
443                         TAILQ_INIT(&sc->bioq);
444                         TAILQ_INIT(&sc->tag_freeq);
445                         TAILQ_INIT(&sc->tag_pendq);
446
447                         lockmgr(&sc->lk, LK_EXCLUSIVE);
448                         RB_INSERT(xa_softc_tree, &xa_device_tree, sc);
449                         TAILQ_INSERT_TAIL(&sc->spanq, msg->state, user_entry);
450                         msg->state->any.xa_sc = sc;
451
452                         /*
453                          * Setup block device
454                          */
455                         for (n = 0; n < MAXTAGS; ++n) {
456                                 tag = kmalloc(sizeof(*tag),
457                                               M_XDISK, M_WAITOK|M_ZERO);
458                                 tag->sc = sc;
459                                 TAILQ_INSERT_TAIL(&sc->tag_freeq, tag, entry);
460                         }
461
462                         if (sc->dev == NULL) {
463                                 dev = disk_create(unit, &sc->disk, &xa_ops);
464                                 dev->si_drv1 = sc;
465                                 sc->dev = dev;
466                                 devstat_add_entry(&sc->stats, "xa", unit,
467                                                   DEV_BSIZE,
468                                                   DEVSTAT_NO_ORDERED_TAGS,
469                                                   DEVSTAT_TYPE_DIRECT |
470                                                   DEVSTAT_TYPE_IF_OTHER,
471                                                   DEVSTAT_PRIORITY_OTHER);
472                         }
473
474                         sc->info.d_media_blksize =
475                                 msg->any.lnk_span.media.block.blksize;
476                         if (sc->info.d_media_blksize <= 0)
477                                 sc->info.d_media_blksize = 1;
478                         sc->info.d_media_blocks =
479                                 msg->any.lnk_span.media.block.bytes /
480                                 sc->info.d_media_blksize;
481                         sc->info.d_dsflags = DSO_MBRQUIET | DSO_RAWPSIZE;
482                         sc->info.d_secpertrack = 32;
483                         sc->info.d_nheads = 64;
484                         sc->info.d_secpercyl = sc->info.d_secpertrack *
485                                                sc->info.d_nheads;
486                         sc->info.d_ncylinders = 0;
487                         if (sc->fs_label[0])
488                                 sc->info.d_serialno = sc->fs_label;
489                         /*
490                          * WARNING! disk_setdiskinfo() must be asynchronous
491                          *          because we are in the rxmsg thread.  If
492                          *          it is synchronous and issues more disk
493                          *          I/Os, we will deadlock.
494                          */
495                         kprintf("xdisk: A1\n");
496                         disk_setdiskinfo(&sc->disk, &sc->info);
497                         xa_restart_deferred(sc);        /* eats serializing */
498                         lockmgr(&sc->lk, LK_RELEASE);
499                 } else {
500                         lockmgr(&sc->lk, LK_EXCLUSIVE);
501                         ++sc->spancnt;
502                         kprintf("xdisk: A2 (%d) ser=%d otag=%p\n", sc->spancnt, sc->serializing, sc->open_tag);
503                         TAILQ_INSERT_TAIL(&sc->spanq, msg->state, user_entry);
504                         msg->state->any.xa_sc = sc;
505                         if (sc->serializing == 0 && sc->open_tag == NULL) {
506                                 sc->serializing = 1;
507                                 xa_restart_deferred(sc); /* eats serializing */
508                         }
509                         lockmgr(&sc->lk, LK_RELEASE);
510                         if (sc->dev && sc->dev->si_disk) {
511                                 kprintf("reprobe\n");
512                                 disk_msg_send(DISK_DISK_REPROBE,
513                                               sc->dev->si_disk,
514                                               NULL);
515                         }
516                 }
517                 kprintf("xdisk: sc %p spancnt %d\n", sc, sc->spancnt);
518                 kdmsg_msg_result(msg, 0);
519                 break;
520         case DMSG_LNK_SPAN | DMSGF_DELETE:
521                 /*
522                  * Manage the tracking node for the remote LNK_SPAN.
523                  *
524                  * Return a final result, closing our end of the transaction.
525                  */
526                 sc = msg->state->any.xa_sc;
527                 kprintf("xdisk: LINK_SPAN state %p delete for %s (sc=%p)\n",
528                         msg->state, (sc ? sc->fs_label : "(null)"), sc);
529                 lockmgr(&sc->lk, LK_EXCLUSIVE);
530                 msg->state->any.xa_sc = NULL;
531                 TAILQ_REMOVE(&sc->spanq, msg->state, user_entry);
532                 --sc->spancnt;
533
534                 kprintf("xdisk: sc %p spancnt %d\n", sc, sc->spancnt);
535
536                 /*
537                  * Spans can come and go as the graph stabilizes, so if
538                  * we lose a span along with sc->open_tag we may be able
539                  * to restart the I/Os on a different span.
540                  */
541                 if (sc->spancnt &&
542                     sc->serializing == 0 && sc->open_tag == NULL) {
543                         sc->serializing = 1;
544                         xa_restart_deferred(sc);
545                 }
546                 lockmgr(&sc->lk, LK_RELEASE);
547                 kdmsg_msg_reply(msg, 0);
548
549 #if 0
550                 /*
551                  * Termination
552                  */
553                 if (sc->spancnt == 0)
554                         xa_terminate_check(sc);
555 #endif
556                 break;
557         case DMSG_LNK_SPAN | DMSGF_DELETE | DMSGF_REPLY:
558                 /*
559                  * Ignore unimplemented streaming replies on our LNK_SPAN
560                  * transaction.
561                  */
562                 kprintf("xdisk: LINK_SPAN state %p delete+reply\n",
563                         msg->state);
564                 break;
565         case DMSG_LNK_SPAN | DMSGF_REPLY:
566                 /*
567                  * Ignore unimplemented streaming replies on our LNK_SPAN
568                  * transaction.
569                  */
570                 kprintf("xdisk: LINK_SPAN state %p reply\n",
571                         msg->state);
572                 break;
573         case DMSG_DBG_SHELL:
574                 /*
575                  * Execute shell command (not supported atm).
576                  *
577                  * This is a one-way packet but if not (e.g. if part of
578                  * a streaming transaction), we will have already closed
579                  * our end.
580                  */
581                 kdmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
582                 break;
583         case DMSG_DBG_SHELL | DMSGF_REPLY:
584                 /*
585                  * Receive one or more replies to a shell command
586                  * that we sent.  Just dump it to the console.
587                  *
588                  * This is a one-way packet but if not (e.g. if
589                  * part of a streaming transaction), we will have
590                  * already closed our end.
591                  */
592                 if (msg->aux_data) {
593                         msg->aux_data[msg->aux_size - 1] = 0;
594                         kprintf("xdisk: DEBUGMSG: %s\n",
595                                 msg->aux_data);
596                 }
597                 break;
598         default:
599                 /*
600                  * Unsupported one-way message, streaming message, or
601                  * transaction.
602                  *
603                  * Terminate any unsupported transactions with an error
604                  * and ignore any unsupported streaming messages.
605                  *
606                  * NOTE: This case also includes DMSG_LNK_ERROR messages
607                  *       which might be one-way, replying to those would
608                  *       cause an infinite ping-pong.
609                  */
610                 if (msg->any.head.cmd & DMSGF_CREATE)
611                         kdmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
612                 break;
613         }
614         lockmgr(&xdisk_lk, LK_RELEASE);
615
616         return 0;
617 }
618
619 /*
620  * Determine if we can destroy the xa_softc.
621  *
622  * Called with xdisk_lk held.
623  */
624 static
625 void
626 xa_terminate_check(struct xa_softc *sc)
627 {
628         xa_tag_t *tag;
629
630         /*
631          * Determine if we can destroy the softc.
632          */
633         kprintf("xdisk: terminate check xa%d (%d,%d,%d) sc=%p ",
634                 sc->unit,
635                 sc->opencnt, sc->serializing, sc->spancnt,
636                 sc);
637
638         if (sc->opencnt || sc->serializing || sc->spancnt) {
639                 kprintf("(leave intact)\n");
640                 return;
641         }
642
643         /*
644          * Remove from device tree, a race with a new incoming span
645          * will create a new softc and disk.
646          */
647         RB_REMOVE(xa_softc_tree, &xa_device_tree, sc);
648
649         /*
650          * Device has to go first to prevent device ops races.
651          */
652         if (sc->dev) {
653                 disk_destroy(&sc->disk);
654                 devstat_remove_entry(&sc->stats);
655                 sc->dev->si_drv1 = NULL;
656                 sc->dev = NULL;
657         }
658
659         kprintf("(remove from tree)\n");
660         sc->serializing = 1;
661         KKASSERT(sc->opencnt == 0);
662         KKASSERT(TAILQ_EMPTY(&sc->tag_pendq));
663
664         while ((tag = TAILQ_FIRST(&sc->tag_freeq)) != NULL) {
665                 TAILQ_REMOVE(&sc->tag_freeq, tag, entry);
666                 tag->sc = NULL;
667                 kfree(tag, M_XDISK);
668         }
669
670         kfree(sc, M_XDISK);
671 }
672
673 /************************************************************************
674  *                         XA DEVICE INTERFACE                          *
675  ************************************************************************/
676
677 static int
678 xa_open(struct dev_open_args *ap)
679 {
680         cdev_t dev = ap->a_head.a_dev;
681         xa_softc_t *sc;
682         int error;
683
684         dev->si_bsize_phys = 512;
685         dev->si_bsize_best = 32768;
686
687         /*
688          * Interlock open with opencnt, wait for attachment operations
689          * to finish.
690          */
691         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
692 again:
693         sc = dev->si_drv1;
694         if (sc == NULL) {
695                 lockmgr(&xdisk_lk, LK_RELEASE);
696                 return ENXIO;   /* raced destruction */
697         }
698         if (sc->serializing) {
699                 tsleep(sc, 0, "xarace", hz / 10);
700                 goto again;
701         }
702         sc->serializing = 1;
703
704         /*
705          * Serialize initial open
706          */
707         if (sc->opencnt++ > 0) {
708                 sc->serializing = 0;
709                 wakeup(sc);
710                 lockmgr(&xdisk_lk, LK_RELEASE);
711                 return(0);
712         }
713
714         /*
715          * Issue BLK_OPEN if necessary.  ENXIO is returned if we have trouble.
716          */
717         if (sc->open_tag == NULL) {
718                 lockmgr(&sc->lk, LK_EXCLUSIVE);
719                 xa_restart_deferred(sc); /* eats serializing */
720                 lockmgr(&sc->lk, LK_RELEASE);
721         } else {
722                 sc->serializing = 0;
723                 wakeup(sc);
724         }
725         lockmgr(&xdisk_lk, LK_RELEASE);
726
727         /*
728          * Wait for completion of the BLK_OPEN
729          */
730         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
731         while (sc->serializing)
732                 lksleep(sc, &xdisk_lk, 0, "xaopen", hz);
733
734         error = sc->last_error;
735         if (error) {
736                 KKASSERT(sc->opencnt > 0);
737                 --sc->opencnt;
738                 xa_terminate_check(sc);
739                 sc = NULL;      /* sc may be invalid now */
740         }
741         lockmgr(&xdisk_lk, LK_RELEASE);
742
743         return (error);
744 }
745
746 static int
747 xa_close(struct dev_close_args *ap)
748 {
749         cdev_t dev = ap->a_head.a_dev;
750         xa_softc_t *sc;
751         xa_tag_t *tag;
752
753         sc = dev->si_drv1;
754         if (sc == NULL)
755                 return ENXIO;   /* raced destruction */
756         lockmgr(&xdisk_lk, LK_EXCLUSIVE);
757         lockmgr(&sc->lk, LK_EXCLUSIVE);
758
759         /*
760          * NOTE: Clearing open_tag allows a concurrent open to re-open
761          *       the device and prevents autonomous completion of the tag.
762          */
763         if (sc->opencnt == 1 && sc->open_tag) {
764                 tag = sc->open_tag;
765                 sc->open_tag = NULL;
766                 lockmgr(&sc->lk, LK_RELEASE);
767                 kdmsg_state_reply(tag->state, 0);       /* close our side */
768                 xa_wait(tag);                           /* wait on remote */
769         } else {
770                 lockmgr(&sc->lk, LK_RELEASE);
771         }
772         KKASSERT(sc->opencnt > 0);
773         --sc->opencnt;
774         xa_terminate_check(sc);
775         lockmgr(&xdisk_lk, LK_RELEASE);
776
777         return(0);
778 }
779
780 static int
781 xa_strategy(struct dev_strategy_args *ap)
782 {
783         xa_softc_t *sc = ap->a_head.a_dev->si_drv1;
784         xa_tag_t *tag;
785         struct bio *bio = ap->a_bio;
786
787         devstat_start_transaction(&sc->stats);
788         atomic_add_int(&xa_active, 1);
789         xa_last = bio->bio_offset;
790
791         /*
792          * If no tags are available NULL is returned and the bio is
793          * placed on sc->bioq.
794          */
795         lockmgr(&sc->lk, LK_EXCLUSIVE);
796         tag = xa_setup_cmd(sc, bio);
797         if (tag)
798                 xa_start(tag, NULL, 1);
799         lockmgr(&sc->lk, LK_RELEASE);
800
801         return(0);
802 }
803
804 static int
805 xa_ioctl(struct dev_ioctl_args *ap)
806 {
807         return(ENOTTY);
808 }
809
810 static int
811 xa_size(struct dev_psize_args *ap)
812 {
813         struct xa_softc *sc;
814
815         if ((sc = ap->a_head.a_dev->si_drv1) == NULL)
816                 return (ENXIO);
817         ap->a_result = sc->info.d_media_blocks;
818         return (0);
819 }
820
821 /************************************************************************
822  *                  XA BLOCK PROTOCOL STATE MACHINE                     *
823  ************************************************************************
824  *
825  * Implement tag/msg setup and related functions.
826  * Called with sc->lk held.
827  */
828 static xa_tag_t *
829 xa_setup_cmd(xa_softc_t *sc, struct bio *bio)
830 {
831         xa_tag_t *tag;
832
833         /*
834          * Only get a tag if we have a valid virtual circuit to the server.
835          */
836         if ((tag = TAILQ_FIRST(&sc->tag_freeq)) != NULL) {
837                 TAILQ_REMOVE(&sc->tag_freeq, tag, entry);
838                 tag->bio = bio;
839                 TAILQ_INSERT_TAIL(&sc->tag_pendq, tag, entry);
840         }
841
842         /*
843          * If we can't dispatch now and this is a bio, queue it for later.
844          */
845         if (tag == NULL && bio) {
846                 TAILQ_INSERT_TAIL(&sc->bioq, bio, bio_act);
847         }
848
849         return (tag);
850 }
851
852 /*
853  * Called with sc->lk held
854  */
855 static void
856 xa_start(xa_tag_t *tag, kdmsg_msg_t *msg, int async)
857 {
858         xa_softc_t *sc = tag->sc;
859
860         tag->done = 0;
861         tag->async = async;
862         tag->status.head.error = DMSG_ERR_IO;   /* fallback error */
863
864         if (msg == NULL) {
865                 struct bio *bio;
866                 struct buf *bp;
867                 kdmsg_state_t *trans;
868
869                 if (sc->opencnt == 0 || sc->open_tag == NULL) {
870                         TAILQ_FOREACH(trans, &sc->spanq, user_entry) {
871                                 if ((trans->rxcmd & DMSGF_DELETE) == 0)
872                                         break;
873                         }
874                 } else {
875                         trans = sc->open_tag->state;
876                 }
877                 if (trans == NULL)
878                         goto skip;
879
880                 KKASSERT(tag->bio);
881                 bio = tag->bio;
882                 bp = bio->bio_buf;
883
884                 switch(bp->b_cmd) {
885                 case BUF_CMD_READ:
886                         msg = kdmsg_msg_alloc(trans,
887                                               DMSG_BLK_READ |
888                                               DMSGF_CREATE |
889                                               DMSGF_DELETE,
890                                               xa_bio_completion, tag);
891                         msg->any.blk_read.keyid = sc->keyid;
892                         msg->any.blk_read.offset = bio->bio_offset;
893                         msg->any.blk_read.bytes = bp->b_bcount;
894                         break;
895                 case BUF_CMD_WRITE:
896                         msg = kdmsg_msg_alloc(trans,
897                                               DMSG_BLK_WRITE |
898                                               DMSGF_CREATE | DMSGF_DELETE,
899                                               xa_bio_completion, tag);
900                         msg->any.blk_write.keyid = sc->keyid;
901                         msg->any.blk_write.offset = bio->bio_offset;
902                         msg->any.blk_write.bytes = bp->b_bcount;
903                         msg->aux_data = bp->b_data;
904                         msg->aux_size = bp->b_bcount;
905                         break;
906                 case BUF_CMD_FLUSH:
907                         msg = kdmsg_msg_alloc(trans,
908                                               DMSG_BLK_FLUSH |
909                                               DMSGF_CREATE | DMSGF_DELETE,
910                                               xa_bio_completion, tag);
911                         msg->any.blk_flush.keyid = sc->keyid;
912                         msg->any.blk_flush.offset = bio->bio_offset;
913                         msg->any.blk_flush.bytes = bp->b_bcount;
914                         break;
915                 case BUF_CMD_FREEBLKS:
916                         msg = kdmsg_msg_alloc(trans,
917                                               DMSG_BLK_FREEBLKS |
918                                               DMSGF_CREATE | DMSGF_DELETE,
919                                               xa_bio_completion, tag);
920                         msg->any.blk_freeblks.keyid = sc->keyid;
921                         msg->any.blk_freeblks.offset = bio->bio_offset;
922                         msg->any.blk_freeblks.bytes = bp->b_bcount;
923                         break;
924                 default:
925                         bp->b_flags |= B_ERROR;
926                         bp->b_error = EIO;
927                         devstat_end_transaction_buf(&sc->stats, bp);
928                         atomic_add_int(&xa_active, -1);
929                         biodone(bio);
930                         tag->bio = NULL;
931                         break;
932                 }
933         }
934
935         /*
936          * If no msg was allocated we likely could not find a good span.
937          */
938 skip:
939         if (msg) {
940                 /*
941                  * Message was passed in or constructed.
942                  */
943                 tag->state = msg->state;
944                 lockmgr(&sc->lk, LK_RELEASE);
945                 kdmsg_msg_write(msg);
946                 lockmgr(&sc->lk, LK_EXCLUSIVE);
947         } else if (tag->bio &&
948                    (tag->bio->bio_buf->b_flags & B_FAILONDIS) == 0) {
949                 /*
950                  * No spans available but BIO is not allowed to fail
951                  * on connectivity problems.  Requeue the BIO.
952                  */
953                 TAILQ_INSERT_TAIL(&sc->bioq, tag->bio, bio_act);
954                 tag->bio = NULL;
955                 lockmgr(&sc->lk, LK_RELEASE);
956                 xa_done(tag, 1);
957                 lockmgr(&sc->lk, LK_EXCLUSIVE);
958         } else {
959                 /*
960                  * No spans available, bio is allowed to fail.
961                  */
962                 lockmgr(&sc->lk, LK_RELEASE);
963                 tag->status.head.error = DMSG_ERR_IO;
964                 xa_done(tag, 1);
965                 lockmgr(&sc->lk, LK_EXCLUSIVE);
966         }
967 }
968
969 static uint32_t
970 xa_wait(xa_tag_t *tag)
971 {
972         xa_softc_t *sc = tag->sc;
973         uint32_t error;
974
975         lockmgr(&sc->lk, LK_EXCLUSIVE);
976         tag->waiting = 1;
977         while (tag->done == 0)
978                 lksleep(tag, &sc->lk, 0, "xawait", 0);
979         lockmgr(&sc->lk, LK_RELEASE);
980
981         error = tag->status.head.error;
982         tag->waiting = 0;
983         xa_release(tag, 0);
984
985         return error;
986 }
987
988 static void
989 xa_done(xa_tag_t *tag, int wasbio)
990 {
991         KKASSERT(tag->bio == NULL);
992
993         tag->state = NULL;
994         tag->done = 1;
995         if (tag->waiting)
996                 wakeup(tag);
997         if (tag->async)
998                 xa_release(tag, wasbio);
999 }
1000
1001 /*
1002  * Release a tag.  If everything looks ok and there are pending BIOs
1003  * (due to all tags in-use), we can use the tag to start the next BIO.
1004  * Do not try to restart if the connection is currently failed.
1005  */
1006 static
1007 void
1008 xa_release(xa_tag_t *tag, int wasbio)
1009 {
1010         xa_softc_t *sc = tag->sc;
1011         struct bio *bio;
1012
1013         if ((bio = tag->bio) != NULL) {
1014                 struct buf *bp = bio->bio_buf;
1015
1016                 bp->b_error = EIO;
1017                 bp->b_flags |= B_ERROR;
1018                 devstat_end_transaction_buf(&sc->stats, bp);
1019                 atomic_add_int(&xa_active, -1);
1020                 biodone(bio);
1021                 tag->bio = NULL;
1022         }
1023
1024         lockmgr(&sc->lk, LK_EXCLUSIVE);
1025
1026         if (wasbio && sc->open_tag &&
1027             (bio = TAILQ_FIRST(&sc->bioq)) != NULL) {
1028                 TAILQ_REMOVE(&sc->bioq, bio, bio_act);
1029                 tag->bio = bio;
1030                 xa_start(tag, NULL, 1);
1031         } else {
1032                 TAILQ_REMOVE(&sc->tag_pendq, tag, entry);
1033                 TAILQ_INSERT_TAIL(&sc->tag_freeq, tag, entry);
1034         }
1035         lockmgr(&sc->lk, LK_RELEASE);
1036 }
1037
1038 /*
1039  * Handle messages under the BLKOPEN transaction.
1040  */
1041 static int
1042 xa_sync_completion(kdmsg_state_t *state, kdmsg_msg_t *msg)
1043 {
1044         xa_tag_t *tag = state->any.any;
1045         xa_softc_t *sc;
1046         struct bio *bio;
1047
1048         /*
1049          * If the tag has been cleaned out we already closed our side
1050          * of the transaction and we are waiting for the other side to
1051          * close.
1052          */
1053         kprintf("xa_sync_completion: tag %p msg %08x state %p\n",
1054                 tag, msg->any.head.cmd, msg->state);
1055
1056         if (tag == NULL) {
1057                 if (msg->any.head.cmd & DMSGF_CREATE)
1058                         kdmsg_state_reply(state, DMSG_ERR_LOSTLINK);
1059                 return 0;
1060         }
1061         sc = tag->sc;
1062
1063         /*
1064          * Validate the tag
1065          */
1066         lockmgr(&sc->lk, LK_EXCLUSIVE);
1067
1068         /*
1069          * Handle initial response to our open and restart any deferred
1070          * BIOs on success.
1071          *
1072          * NOTE: DELETE may also be set.
1073          */
1074         if (msg->any.head.cmd & DMSGF_CREATE) {
1075                 switch(msg->any.head.cmd & DMSGF_CMDSWMASK) {
1076                 case DMSG_LNK_ERROR | DMSGF_REPLY:
1077                         bzero(&tag->status, sizeof(tag->status));
1078                         tag->status.head = msg->any.head;
1079                         break;
1080                 case DMSG_BLK_ERROR | DMSGF_REPLY:
1081                         tag->status = msg->any.blk_error;
1082                         break;
1083                 }
1084                 sc->last_error = tag->status.head.error;
1085                 kprintf("xdisk: blk_open completion status %d\n",
1086                         sc->last_error);
1087                 if (sc->last_error == 0) {
1088                         while ((bio = TAILQ_FIRST(&sc->bioq)) != NULL) {
1089                                 tag = xa_setup_cmd(sc, NULL);
1090                                 if (tag == NULL)
1091                                         break;
1092                                 TAILQ_REMOVE(&sc->bioq, bio, bio_act);
1093                                 tag->bio = bio;
1094                                 xa_start(tag, NULL, 1);
1095                         }
1096                 }
1097                 sc->serializing = 0;
1098                 wakeup(sc);
1099         }
1100
1101         /*
1102          * Handle unexpected termination (or lost comm channel) from other
1103          * side.  Autonomous completion only if open_tag matches,
1104          * otherwise another thread is probably waiting on the tag.
1105          *
1106          * (see xa_close() for other interactions)
1107          */
1108         if (msg->any.head.cmd & DMSGF_DELETE) {
1109                 kdmsg_state_reply(tag->state, 0);
1110                 if (sc->open_tag == tag) {
1111                         sc->open_tag = NULL;
1112                         xa_done(tag, 0);
1113                 } else {
1114                         tag->async = 0;
1115                         xa_done(tag, 0);
1116                 }
1117         }
1118         lockmgr(&sc->lk, LK_RELEASE);
1119
1120         return (0);
1121 }
1122
1123 static int
1124 xa_bio_completion(kdmsg_state_t *state, kdmsg_msg_t *msg)
1125 {
1126         xa_tag_t *tag = state->any.any;
1127         xa_softc_t *sc = tag->sc;
1128         struct bio *bio;
1129         struct buf *bp;
1130
1131         /*
1132          * Get the bio from the tag.  If no bio is present we just do
1133          * 'done' handling.
1134          */
1135         if ((bio = tag->bio) == NULL)
1136                 goto handle_done;
1137         bp = bio->bio_buf;
1138
1139         /*
1140          * Process return status
1141          */
1142         switch(msg->any.head.cmd & DMSGF_CMDSWMASK) {
1143         case DMSG_LNK_ERROR | DMSGF_REPLY:
1144                 bzero(&tag->status, sizeof(tag->status));
1145                 tag->status.head = msg->any.head;
1146                 if (tag->status.head.error)
1147                         tag->status.resid = bp->b_bcount;
1148                 else
1149                         tag->status.resid = 0;
1150                 break;
1151         case DMSG_BLK_ERROR | DMSGF_REPLY:
1152                 tag->status = msg->any.blk_error;
1153                 break;
1154         }
1155
1156         /*
1157          * If the device is open stall the bio on DMSG errors.  If an
1158          * actual I/O error occured on the remote device, DMSG_ERR_IO
1159          * will be returned.
1160          */
1161         if (tag->status.head.error &&
1162             (msg->any.head.cmd & DMSGF_DELETE) && sc->opencnt) {
1163                 if (tag->status.head.error != DMSG_ERR_IO)
1164                         goto handle_repend;
1165         }
1166
1167         /*
1168          * Process bio completion
1169          *
1170          * For reads any returned data is zero-extended if necessary, so
1171          * the server can short-cut any all-zeros reads if it desires.
1172          */
1173         switch(bp->b_cmd) {
1174         case BUF_CMD_READ:
1175                 if (msg->aux_data && msg->aux_size) {
1176                         if (msg->aux_size < bp->b_bcount) {
1177                                 bcopy(msg->aux_data, bp->b_data, msg->aux_size);
1178                                 bzero(bp->b_data + msg->aux_size,
1179                                       bp->b_bcount - msg->aux_size);
1180                         } else {
1181                                 bcopy(msg->aux_data, bp->b_data, bp->b_bcount);
1182                         }
1183                 } else {
1184                         bzero(bp->b_data, bp->b_bcount);
1185                 }
1186                 /* fall through */
1187         case BUF_CMD_WRITE:
1188         case BUF_CMD_FLUSH:
1189         case BUF_CMD_FREEBLKS:
1190         default:
1191                 if (tag->status.resid > bp->b_bcount)
1192                         tag->status.resid = bp->b_bcount;
1193                 bp->b_resid = tag->status.resid;
1194                 if (tag->status.head.error != 0) {
1195                         bp->b_error = EIO;
1196                         bp->b_flags |= B_ERROR;
1197                 } else {
1198                         bp->b_resid = 0;
1199                 }
1200                 devstat_end_transaction_buf(&sc->stats, bp);
1201                 atomic_add_int(&xa_active, -1);
1202                 biodone(bio);
1203                 tag->bio = NULL;
1204                 break;
1205         }
1206
1207         /*
1208          * Handle completion of the transaction.  If the bioq is not empty
1209          * we can initiate another bio on the same tag.
1210          *
1211          * NOTE: Most of our transactions will be single-message
1212          *       CREATE+DELETEs, so we won't have to terminate the
1213          *       transaction separately, here.  But just in case they
1214          *       aren't be sure to terminate the transaction.
1215          */
1216 handle_done:
1217         if (msg->any.head.cmd & DMSGF_DELETE) {
1218                 xa_done(tag, 1);
1219                 if ((state->txcmd & DMSGF_DELETE) == 0)
1220                         kdmsg_msg_reply(msg, 0);
1221         }
1222         return (0);
1223
1224         /*
1225          * Handle the case where the transaction failed due to a
1226          * connectivity issue.  The tag is put away with wasbio=0
1227          * and we put the BIO back onto the bioq for a later restart.
1228          *
1229          * probe I/Os (where the device is not open) will be failed
1230          * instead of requeued.
1231          */
1232 handle_repend:
1233         tag->bio = NULL;
1234         if (bio->bio_buf->b_flags & B_FAILONDIS) {
1235                 kprintf("xa_strategy: disconnected, fail bp %p\n",
1236                         bio->bio_buf);
1237                 bio->bio_buf->b_error = ENXIO;
1238                 bio->bio_buf->b_flags |= B_ERROR;
1239                 biodone(bio);
1240                 bio = NULL;
1241                 kprintf("BIO CIRC FAILURE, FAIL BIO %p\n", bio);
1242         } else {
1243                 kprintf("BIO CIRC FAILURE, REPEND BIO %p\n", bio);
1244         }
1245         xa_done(tag, 0);
1246         if ((state->txcmd & DMSGF_DELETE) == 0)
1247                 kdmsg_msg_reply(msg, 0);
1248
1249         /*
1250          * Requeue the bio
1251          */
1252         if (bio) {
1253                 lockmgr(&sc->lk, LK_EXCLUSIVE);
1254                 TAILQ_INSERT_TAIL(&sc->bioq, bio, bio_act);
1255                 lockmgr(&sc->lk, LK_RELEASE);
1256         }
1257         return (0);
1258 }
1259
1260 /*
1261  * Restart as much deferred I/O as we can.  The serializer is set and we
1262  * eat it (clear it) when done.
1263  *
1264  * Called with sc->lk held
1265  */
1266 static
1267 void
1268 xa_restart_deferred(xa_softc_t *sc)
1269 {
1270         kdmsg_state_t *span;
1271         kdmsg_msg_t *msg;
1272         xa_tag_t *tag;
1273         int error;
1274
1275         KKASSERT(sc->serializing);
1276
1277         /*
1278          * Determine if a restart is needed.
1279          */
1280         if (sc->opencnt == 0) {
1281                 /*
1282                  * Device is not open, nothing to do, eat serializing.
1283                  */
1284                 sc->serializing = 0;
1285                 wakeup(sc);
1286         } else if (sc->open_tag == NULL) {
1287                 /*
1288                  * BLK_OPEN required before we can restart any BIOs.
1289                  * Select the best LNK_SPAN to issue the BLK_OPEN under.
1290                  *
1291                  * serializing interlocks waiting open()s.
1292                  */
1293                 error = 0;
1294                 TAILQ_FOREACH(span, &sc->spanq, user_entry) {
1295                         if ((span->rxcmd & DMSGF_DELETE) == 0)
1296                                 break;
1297                 }
1298                 if (span == NULL)
1299                         error = ENXIO;
1300
1301                 if (error == 0) {
1302                         tag = xa_setup_cmd(sc, NULL);
1303                         if (tag == NULL)
1304                                 error = ENXIO;
1305                 }
1306                 if (error == 0) {
1307                         sc->open_tag = tag;
1308                         msg = kdmsg_msg_alloc(span,
1309                                               DMSG_BLK_OPEN |
1310                                               DMSGF_CREATE,
1311                                               xa_sync_completion, tag);
1312                         msg->any.blk_open.modes = DMSG_BLKOPEN_RD;
1313                         kprintf("xdisk: BLK_OPEN tag %p state %p "
1314                                 "span-state %p\n",
1315                                 tag, msg->state, span);
1316                         xa_start(tag, msg, 0);
1317                 }
1318                 if (error) {
1319                         sc->serializing = 0;
1320                         wakeup(sc);
1321                 }
1322                 /* else leave serializing set until BLK_OPEN response */
1323         } else {
1324                 /* nothing to do */
1325                 sc->serializing = 0;
1326                 wakeup(sc);
1327         }
1328 }