sys/kern/kern_syslink.c

   1 /*
   2  * Copyright (c) 2006-2007 The DragonFly Project.  All rights reserved.
   3  *
   4  * This code is derived from software contributed to The DragonFly Project
   5  * by Matthew Dillon <dillon@backplane.com>
   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  * $DragonFly: src/sys/kern/kern_syslink.c,v 1.16 2008/10/26 04:29:19 sephe Exp $
  35  */
  36 /*
  37  * This module implements the core syslink() system call and provides
  38  * glue for kernel syslink frontends and backends, creating a intra-host
  39  * communications infrastructure and DMA transport abstraction.
  40  */
  41
  42 #include <sys/param.h>
  43 #include <sys/systm.h>
  44 #include <sys/kernel.h>
  45 #include <sys/endian.h>
  46 #include <sys/malloc.h>
  47 #include <sys/alist.h>
  48 #include <sys/file.h>
  49 #include <sys/proc.h>
  50 #include <sys/priv.h>
  51 #include <sys/lock.h>
  52 #include <sys/uio.h>
  53 #include <sys/objcache.h>
  54 #include <sys/queue.h>
  55 #include <sys/thread.h>
  56 #include <sys/tree.h>
  57 #include <sys/sysctl.h>
  58 #include <sys/sysproto.h>
  59 #include <sys/mbuf.h>
  60 #include <sys/socket.h>
  61 #include <sys/socketvar.h>
  62 #include <sys/socketops.h>
  63 #include <sys/sysref.h>
  64 #include <sys/syslink.h>
  65 #include <sys/syslink_msg.h>
  66 #include <netinet/in.h>
  67
  68 #include <sys/thread2.h>
  69 #include <sys/spinlock2.h>
  70 #include <sys/buf2.h>
  71 #include <sys/mplock2.h>
  72
  73 #include "opt_syslink.h"
  74
  75 /*
  76  * Syslink Connection abstraction
  77  */
  78 struct slcommon {
  79         struct spinlock spin;
  80         int     refs;
  81 };
  82
  83 struct sldesc {
  84         struct slmsgq   inq;
  85         struct slmsg_rb_tree reply_rb_root; /* replies to requests */
  86         struct spinlock spin;
  87         struct sldesc   *peer;          /* peer syslink, if any */
  88         struct file     *xfp;           /* external file pointer */
  89         struct slcommon *common;
  90         int     flags;
  91         int     rwaiters;               /* number of threads waiting */
  92         int     wblocked;               /* blocked waiting for us to drain */
  93         size_t  cmdbytes;               /* unreplied commands pending */
  94         size_t  repbytes;               /* undrained replies pending */
  95         int     (*backend_wblocked)(struct sldesc *, int, sl_proto_t);
  96         int     (*backend_write)(struct sldesc *, struct slmsg *);
  97         void    (*backend_reply)(struct sldesc *,struct slmsg *,struct slmsg *);
  98         void    (*backend_dispose)(struct sldesc *, struct slmsg *);
  99 };
 100
 101 #define SLF_RSHUTDOWN   0x0001
 102 #define SLF_WSHUTDOWN   0x0002
 103
 104 static int syslink_cmd_new(struct syslink_info_new *info, int *result);
 105 static struct sldesc *allocsldesc(struct slcommon *common);
 106 static void setsldescfp(struct sldesc *sl, struct file *fp);
 107 static void shutdownsldesc(struct sldesc *sl, int how);
 108 static void shutdownsldesc2(struct sldesc *sl, int how);
 109 static void sldrop(struct sldesc *sl);
 110 static int syslink_validate_msg(struct syslink_msg *msg, int bytes);
 111 static int syslink_validate_elm(struct syslink_elm *elm, sl_reclen_t bytes,
 112                                  int swapit, int depth);
 113
 114 static int sl_local_mmap(struct slmsg *slmsg, char *base, size_t len);
 115 static void sl_local_munmap(struct slmsg *slmsg);
 116
 117 static int backend_wblocked_user(struct sldesc *sl, int nbio, sl_proto_t proto);
 118 static int backend_write_user(struct sldesc *sl, struct slmsg *slmsg);
 119 static void backend_reply_user(struct sldesc *sl, struct slmsg *slcmd,
 120                                struct slmsg *slrep);
 121 static void backend_dispose_user(struct sldesc *sl, struct slmsg *slmsg);
 122
 123 static int backend_wblocked_kern(struct sldesc *sl, int nbio, sl_proto_t proto);
 124 static int backend_write_kern(struct sldesc *sl, struct slmsg *slmsg);
 125 static void backend_reply_kern(struct sldesc *sl, struct slmsg *slcmd,
 126                                struct slmsg *slrep);
 127 static void backend_dispose_kern(struct sldesc *sl, struct slmsg *slmsg);
 128 static void slmsg_put(struct slmsg *slmsg);
 129
 130 /*
 131  * Objcache memory backend
 132  *
 133  * All three object caches return slmsg structures but each is optimized
 134  * for syslink message buffers of varying sizes.  We use the slightly
 135  * more complex ctor/dtor API in order to provide ready-to-go slmsg's.
 136  */
 137
 138 static struct objcache *sl_objcache_big;
 139 static struct objcache *sl_objcache_small;
 140 static struct objcache *sl_objcache_none;
 141
 142 MALLOC_DEFINE(M_SYSLINK, "syslink", "syslink manager");
 143
 144 static boolean_t slmsg_ctor(void *data, void *private, int ocflags);
 145 static void slmsg_dtor(void *data, void *private);
 146
 147 static
 148 void
 149 syslinkinit(void *dummy __unused)
 150 {
 151         size_t n = sizeof(struct slmsg);
 152
 153         sl_objcache_none = objcache_create_mbacked(M_SYSLINK, n, NULL, 64,
 154                                                    slmsg_ctor, slmsg_dtor,
 155                                                    &sl_objcache_none);
 156         sl_objcache_small= objcache_create_mbacked(M_SYSLINK, n, NULL, 64,
 157                                                    slmsg_ctor, slmsg_dtor,
 158                                                    &sl_objcache_small);
 159         sl_objcache_big  = objcache_create_mbacked(M_SYSLINK, n, NULL, 16,
 160                                                    slmsg_ctor, slmsg_dtor,
 161                                                    &sl_objcache_big);
 162 }
 163
 164 static
 165 boolean_t
 166 slmsg_ctor(void *data, void *private, int ocflags)
 167 {
 168         struct slmsg *slmsg = data;
 169
 170         bzero(slmsg, sizeof(*slmsg));
 171
 172         slmsg->oc = *(struct objcache **)private;
 173         if (slmsg->oc == sl_objcache_none) {
 174                 slmsg->maxsize = 0;
 175         } else if (slmsg->oc == sl_objcache_small) {
 176                 slmsg->maxsize = SLMSG_SMALL;
 177         } else if (slmsg->oc == sl_objcache_big) {
 178                 slmsg->maxsize = SLMSG_BIG;
 179         } else {
 180                 panic("slmsg_ctor: bad objcache?\n");
 181         }
 182         if (slmsg->maxsize) {
 183                 slmsg->msg = kmalloc(slmsg->maxsize,
 184                                      M_SYSLINK, M_WAITOK|M_ZERO);
 185         }
 186         xio_init(&slmsg->xio);
 187         return(TRUE);
 188 }
 189
 190 static
 191 void
 192 slmsg_dtor(void *data, void *private)
 193 {
 194         struct slmsg *slmsg = data;
 195
 196         if (slmsg->maxsize && slmsg->msg) {
 197                 kfree(slmsg->msg, M_SYSLINK);
 198                 slmsg->msg = NULL;
 199         }
 200         slmsg->oc = NULL;
 201 }
 202
 203 SYSINIT(syslink, SI_BOOT2_MACHDEP, SI_ORDER_ANY, syslinkinit, NULL)
 204
 205 static int rb_slmsg_compare(struct slmsg *msg1, struct slmsg *msg2);
 206 RB_GENERATE2(slmsg_rb_tree, slmsg, rbnode, rb_slmsg_compare,
 207              sysid_t, msg->sm_msgid);
 208
 209 /*
 210  * Sysctl elements
 211  */
 212 static int syslink_enabled;
 213 SYSCTL_NODE(_kern, OID_AUTO, syslink, CTLFLAG_RW, 0, "Pipe operation");
 214 SYSCTL_INT(_kern_syslink, OID_AUTO, enabled,
 215             CTLFLAG_RW, &syslink_enabled, 0, "Enable SYSLINK");
 216 static size_t syslink_bufsize = 65536;
 217 SYSCTL_UINT(_kern_syslink, OID_AUTO, bufsize,
 218             CTLFLAG_RW, &syslink_bufsize, 0, "Maximum buffer size");
 219
 220 /*
 221  * Fileops API - typically used to glue a userland frontend with a
 222  *               kernel backend.
 223  */
 224
 225 static int slfileop_read(struct file *fp, struct uio *uio,
 226                         struct ucred *cred, int flags);
 227 static int slfileop_write(struct file *fp, struct uio *uio,
 228                          struct ucred *cred, int flags);
 229 static int slfileop_close(struct file *fp);
 230 static int slfileop_stat(struct file *fp, struct stat *sb, struct ucred *cred);
 231 static int slfileop_shutdown(struct file *fp, int how);
 232 static int slfileop_ioctl(struct file *fp, u_long cmd, caddr_t data,
 233                          struct ucred *cred, struct sysmsg *msg);
 234 static int slfileop_poll(struct file *fp, int events, struct ucred *cred);
 235 static int slfileop_kqfilter(struct file *fp, struct knote *kn);
 236
 237 static struct fileops syslinkops = {
 238     .fo_read =          slfileop_read,
 239     .fo_write =         slfileop_write,
 240     .fo_ioctl =         slfileop_ioctl,
 241     .fo_poll =          slfileop_poll,
 242     .fo_kqfilter =      slfileop_kqfilter,
 243     .fo_stat =          slfileop_stat,
 244     .fo_close =         slfileop_close,
 245     .fo_shutdown =      slfileop_shutdown
 246 };
 247
 248 /************************************************************************
 249  *                      PRIMARY SYSTEM CALL INTERFACE                   *
 250  ************************************************************************
 251  *
 252  * syslink(int cmd, struct syslink_info *info, size_t bytes)
 253  *
 254  * MPALMOSTSAFE
 255  */
 256 int
 257 sys_syslink(struct syslink_args *uap)
 258 {
 259         union syslink_info_all info;
 260         int error;
 261
 262         /*
 263          * System call is under construction and disabled by default.
 264          * Superuser access is also required for now, but eventually
 265          * will not be needed.
 266          */
 267         if (syslink_enabled == 0)
 268                 return (EAUTH);
 269         error = priv_check(curthread, PRIV_ROOT);
 270         if (error)
 271                 return (error);
 272
 273         /*
 274          * Load and validate the info structure.  Unloaded bytes are zerod
 275          * out.  The label field must always be 0-filled, even if not used
 276          * for a command.
 277          */
 278         bzero(&info, sizeof(info));
 279         if ((unsigned)uap->bytes <= sizeof(info)) {
 280                 if (uap->bytes)
 281                         error = copyin(uap->info, &info, uap->bytes);
 282         } else {
 283                 error = EINVAL;
 284         }
 285         if (error)
 286                 return (error);
 287         get_mplock();
 288
 289         /*
 290          * Process the command
 291          */
 292         switch(uap->cmd) {
 293         case SYSLINK_CMD_NEW:
 294                 error = syslink_cmd_new(&info.cmd_new, &uap->sysmsg_result);
 295                 break;
 296         default:
 297                 error = EINVAL;
 298                 break;
 299         }
 300
 301         rel_mplock();
 302         if (error == 0 && info.head.wbflag)
 303                 copyout(&info, uap->info, uap->bytes);
 304         return (error);
 305 }
 306
 307 /*
 308  * Create a linked pair of descriptors, like a pipe.
 309  */
 310 static
 311 int
 312 syslink_cmd_new(struct syslink_info_new *info, int *result)
 313 {
 314         struct thread *td = curthread;
 315         struct filedesc *fdp = td->td_proc->p_fd;
 316         struct file *fp1;
 317         struct file *fp2;
 318         struct sldesc *sl;
 319         struct sldesc *slpeer;
 320         int error;
 321         int fd1, fd2;
 322
 323         error = falloc(td->td_lwp, &fp1, &fd1);
 324         if (error)
 325                 return(error);
 326         error = falloc(td->td_lwp, &fp2, &fd2);
 327         if (error) {
 328                 fsetfd(fdp, NULL, fd1);
 329                 fdrop(fp1);
 330                 return(error);
 331         }
 332         slpeer = allocsldesc(NULL);
 333         slpeer->backend_wblocked = backend_wblocked_user;
 334         slpeer->backend_write = backend_write_user;
 335         slpeer->backend_reply = backend_reply_user;
 336         slpeer->backend_dispose = backend_dispose_user;
 337         sl = allocsldesc(slpeer->common);
 338         sl->peer = slpeer;
 339         sl->backend_wblocked = backend_wblocked_user;
 340         sl->backend_write = backend_write_user;
 341         sl->backend_reply = backend_reply_user;
 342         sl->backend_dispose = backend_dispose_user;
 343         slpeer->peer = sl;
 344
 345         setsldescfp(sl, fp1);
 346         setsldescfp(slpeer, fp2);
 347
 348         fsetfd(fdp, fp1, fd1);
 349         fdrop(fp1);
 350         fsetfd(fdp, fp2, fd2);
 351         fdrop(fp2);
 352
 353         info->head.wbflag = 1;  /* write back */
 354         info->fds[0] = fd1;
 355         info->fds[1] = fd2;
 356
 357         return(0);
 358 }
 359
 360 /************************************************************************
 361  *                      LOW LEVEL SLDESC SUPPORT                        *
 362  ************************************************************************
 363  *
 364  */
 365
 366 static
 367 struct sldesc *
 368 allocsldesc(struct slcommon *common)
 369 {
 370         struct sldesc *sl;
 371
 372         sl = kmalloc(sizeof(struct sldesc), M_SYSLINK, M_WAITOK|M_ZERO);
 373         if (common == NULL)
 374                 common = kmalloc(sizeof(*common), M_SYSLINK, M_WAITOK|M_ZERO);
 375         TAILQ_INIT(&sl->inq);           /* incoming requests */
 376         RB_INIT(&sl->reply_rb_root);    /* match incoming replies */
 377         spin_init(&sl->spin);
 378         sl->common = common;
 379         ++common->refs;
 380         return(sl);
 381 }
 382
 383 static
 384 void
 385 setsldescfp(struct sldesc *sl, struct file *fp)
 386 {
 387         sl->xfp = fp;
 388         fp->f_type = DTYPE_SYSLINK;
 389         fp->f_flag = FREAD | FWRITE;
 390         fp->f_ops = &syslinkops;
 391         fp->f_data = sl;
 392 }
 393
 394 /*
 395  * Red-black tree compare function
 396  */
 397 static
 398 int
 399 rb_slmsg_compare(struct slmsg *msg1, struct slmsg *msg2)
 400 {
 401         if (msg1->msg->sm_msgid < msg2->msg->sm_msgid)
 402                 return(-1);
 403         if (msg1->msg->sm_msgid == msg2->msg->sm_msgid)
 404                 return(0);
 405         return(1);
 406 }
 407
 408 static
 409 void
 410 shutdownsldesc(struct sldesc *sl, int how)
 411 {
 412         struct slmsg *slmsg;
 413         int rhow;
 414
 415         shutdownsldesc2(sl, how);
 416
 417         /*
 418          * Return unread and unreplied messages
 419          */
 420         spin_lock_wr(&sl->spin);
 421         while ((slmsg = TAILQ_FIRST(&sl->inq)) != NULL) {
 422                 TAILQ_REMOVE(&sl->inq, slmsg, tqnode);
 423                 spin_unlock_wr(&sl->spin);
 424                 if (slmsg->msg->sm_proto & SM_PROTO_REPLY) {
 425                         sl->repbytes -= slmsg->maxsize;
 426                         slmsg->flags &= ~SLMSGF_ONINQ;
 427                         sl->peer->backend_dispose(sl->peer, slmsg);
 428                 }
 429                 /* leave ONINQ set for commands, it will cleared below */
 430                 spin_lock_wr(&sl->spin);
 431         }
 432         while ((slmsg = RB_ROOT(&sl->reply_rb_root)) != NULL) {
 433                 RB_REMOVE(slmsg_rb_tree, &sl->reply_rb_root, slmsg);
 434                 sl->cmdbytes -= slmsg->maxsize;
 435                 spin_unlock_wr(&sl->spin);
 436                 slmsg->flags &= ~SLMSGF_ONINQ;
 437                 sl->peer->backend_reply(sl->peer, slmsg, NULL);
 438                 spin_lock_wr(&sl->spin);
 439         }
 440         spin_unlock_wr(&sl->spin);
 441
 442         /*
 443          * Call shutdown on the peer with the opposite flags
 444          */
 445         rhow = 0;
 446         switch(how) {
 447         case SHUT_RD:
 448                 rhow = SHUT_WR;
 449                 break;
 450         case SHUT_WR:
 451                 rhow = SHUT_WR;
 452                 break;
 453         case SHUT_RDWR:
 454                 rhow = SHUT_RDWR;
 455                 break;
 456         }
 457         shutdownsldesc2(sl->peer, rhow);
 458 }
 459
 460 static
 461 void
 462 shutdownsldesc2(struct sldesc *sl, int how)
 463 {
 464         spin_lock_wr(&sl->spin);
 465         switch(how) {
 466         case SHUT_RD:
 467                 sl->flags |= SLF_RSHUTDOWN;
 468                 break;
 469         case SHUT_WR:
 470                 sl->flags |= SLF_WSHUTDOWN;
 471                 break;
 472         case SHUT_RDWR:
 473                 sl->flags |= SLF_RSHUTDOWN | SLF_WSHUTDOWN;
 474                 break;
 475         }
 476         spin_unlock_wr(&sl->spin);
 477
 478         /*
 479          * Handle signaling on the user side
 480          */
 481         if (how & SHUT_RD) {
 482                 if (sl->rwaiters)
 483                         wakeup(&sl->rwaiters);
 484         }
 485         if (how & SHUT_WR) {
 486                 if (sl->wblocked) {
 487                         sl->wblocked = 0;       /* race ok */
 488                         wakeup(&sl->wblocked);
 489                 }
 490         }
 491 }
 492
 493 static
 494 void
 495 sldrop(struct sldesc *sl)
 496 {
 497         struct sldesc *slpeer;
 498
 499         spin_lock_wr(&sl->common->spin);
 500         if (--sl->common->refs == 0) {
 501                 spin_unlock_wr(&sl->common->spin);
 502                 if ((slpeer = sl->peer) != NULL) {
 503                         sl->peer = NULL;
 504                         slpeer->peer = NULL;
 505                         slpeer->common = NULL;
 506                         KKASSERT(slpeer->xfp == NULL);
 507                         KKASSERT(TAILQ_EMPTY(&slpeer->inq));
 508                         KKASSERT(RB_EMPTY(&slpeer->reply_rb_root));
 509                         kfree(slpeer, M_SYSLINK);
 510                 }
 511                 KKASSERT(sl->xfp == NULL);
 512                 KKASSERT(TAILQ_EMPTY(&sl->inq));
 513                 KKASSERT(RB_EMPTY(&sl->reply_rb_root));
 514                 kfree(sl->common, M_SYSLINK);
 515                 sl->common = NULL;
 516                 kfree(sl, M_SYSLINK);
 517         } else {
 518                 spin_unlock_wr(&sl->common->spin);
 519         }
 520 }
 521
 522 static
 523 void
 524 slmsg_put(struct slmsg *slmsg)
 525 {
 526         if (slmsg->flags & SLMSGF_HASXIO) {
 527                 slmsg->flags &= ~SLMSGF_HASXIO;
 528                 get_mplock();
 529                 xio_release(&slmsg->xio);
 530                 rel_mplock();
 531         }
 532         slmsg->flags &= ~SLMSGF_LINMAP;
 533         objcache_put(slmsg->oc, slmsg);
 534 }
 535
 536 /************************************************************************
 537  *                              FILEOPS API                             *
 538  ************************************************************************
 539  *
 540  * Implement userland fileops.
 541  *
 542  * MPSAFE ops
 543  */
 544 static
 545 int
 546 slfileop_read(struct file *fp, struct uio *uio, struct ucred *cred, int flags)
 547 {
 548         struct sldesc *sl = fp->f_data;         /* fp refed on call */
 549         struct slmsg *slmsg;
 550         struct iovec *iov0;
 551         struct iovec *iov1;
 552         struct syslink_msg *wmsg;
 553         int error;
 554         int nbio;
 555
 556         /*
 557          * Kinda messy.  Figure out the non-blocking state
 558          */
 559         if (flags & O_FBLOCKING)
 560                 nbio = 0;
 561         else if (flags & O_FNONBLOCKING)
 562                 nbio = 1;
 563         else if (fp->f_flag & O_NONBLOCK)
 564                 nbio = 1;
 565         else
 566                 nbio = 0;
 567
 568         /*
 569          * Validate the uio.
 570          *
 571          * iov0 - message buffer
 572          * iov1 - DMA buffer or backup buffer
 573          */
 574         if (uio->uio_iovcnt < 1) {
 575                 error = 0;
 576                 goto done2;
 577         }
 578         iov0 = &uio->uio_iov[0];
 579         if (uio->uio_iovcnt > 2) {
 580                 error = EINVAL;
 581                 goto done2;
 582         }
 583
 584         /*
 585          * Get a message, blocking if necessary.
 586          */
 587         spin_lock_wr(&sl->spin);
 588         while ((slmsg = TAILQ_FIRST(&sl->inq)) == NULL) {
 589                 if (sl->flags & SLF_RSHUTDOWN) {
 590                         error = 0;
 591                         goto done1;
 592                 }
 593                 if (nbio) {
 594                         error = EAGAIN;
 595                         goto done1;
 596                 }
 597                 ++sl->rwaiters;
 598                 error = ssleep(&sl->rwaiters, &sl->spin, PCATCH, "slrmsg", 0);
 599                 --sl->rwaiters;
 600                 if (error)
 601                         goto done1;
 602         }
 603         wmsg = slmsg->msg;
 604
 605         /*
 606          * We have a message and still hold the spinlock.  Make sure the
 607          * uio has enough room to hold the message.
 608          *
 609          * Note that replies do not have XIOs.
 610          */
 611         if (slmsg->msgsize > iov0->iov_len) {
 612                 error = ENOSPC;
 613                 goto done1;
 614         }
 615         if (slmsg->xio.xio_bytes) {
 616                 if (uio->uio_iovcnt != 2) {
 617                         error = ENOSPC;
 618                         goto done1;
 619                 }
 620                 iov1 = &uio->uio_iov[1];
 621                 if (slmsg->xio.xio_bytes > iov1->iov_len) {
 622                         error = ENOSPC;
 623                         goto done1;
 624                 }
 625         } else {
 626                 iov1 = NULL;
 627         }
 628
 629         /*
 630          * Dequeue the message.  Adjust repbytes immediately.  cmdbytes
 631          * are adjusted when the command is replied to, not here.
 632          */
 633         TAILQ_REMOVE(&sl->inq, slmsg, tqnode);
 634         if (slmsg->msg->sm_proto & SM_PROTO_REPLY)
 635                 sl->repbytes -= slmsg->maxsize;
 636         spin_unlock_wr(&sl->spin);
 637
 638         /*
 639          * Load the message data into the user buffer.
 640          *
 641          * If receiving a command an XIO may exist specifying a DMA buffer.
 642          * For commands, if DMAW is set we have to copy or map the buffer
 643          * so the caller can access the data being written.  If DMAR is set
 644          * we do not have to copy but we still must map the buffer so the
 645          * caller can directly fill in the data being requested.
 646          */
 647         error = uiomove((void *)slmsg->msg, slmsg->msgsize, uio);
 648         if (error == 0 && slmsg->xio.xio_bytes &&
 649             (wmsg->sm_head.se_cmd & SE_CMDF_REPLY) == 0) {
 650                 if (wmsg->sm_head.se_cmd & SE_CMDF_DMAW) {
 651                         /*
 652                          * Data being passed to caller or being passed in both
 653                          * directions, copy or map.
 654                          */
 655                         get_mplock();
 656                         if ((flags & O_MAPONREAD) &&
 657                             (slmsg->xio.xio_flags & XIOF_VMLINEAR)) {
 658                                 error = sl_local_mmap(slmsg,
 659                                                       iov1->iov_base,
 660                                                       iov1->iov_len);
 661                                 if (error)
 662                                 error = xio_copy_xtou(&slmsg->xio, 0,
 663                                                       iov1->iov_base,
 664                                                       slmsg->xio.xio_bytes);
 665                         } else {
 666                                 error = xio_copy_xtou(&slmsg->xio, 0,
 667                                                       iov1->iov_base,
 668                                                       slmsg->xio.xio_bytes);
 669                         }
 670                         rel_mplock();
 671                 } else if (wmsg->sm_head.se_cmd & SE_CMDF_DMAR) {
 672                         /*
 673                          * Data will be passed back to originator, map
 674                          * the buffer if we can, else use the backup
 675                          * buffer at the same VA supplied by the caller.
 676                          */
 677                         get_mplock();
 678                         if ((flags & O_MAPONREAD) &&
 679                             (slmsg->xio.xio_flags & XIOF_VMLINEAR)) {
 680                                 error = sl_local_mmap(slmsg,
 681                                                       iov1->iov_base,
 682                                                       iov1->iov_len);
 683                                 error = 0; /* ignore errors */
 684                         }
 685                         rel_mplock();
 686                 }
 687         }
 688
 689         /*
 690          * Clean up.
 691          */
 692         if (error) {
 693                 /*
 694                  * Requeue the message if we could not read it successfully
 695                  */
 696                 spin_lock_wr(&sl->spin);
 697                 TAILQ_INSERT_HEAD(&sl->inq, slmsg, tqnode);
 698                 slmsg->flags |= SLMSGF_ONINQ;
 699                 spin_unlock_wr(&sl->spin);
 700         } else if (slmsg->msg->sm_proto & SM_PROTO_REPLY) {
 701                 /*
 702                  * Dispose of any received reply after we've copied it
 703                  * to userland.  We don't need the slmsg any more.
 704                  */
 705                 slmsg->flags &= ~SLMSGF_ONINQ;
 706                 sl->peer->backend_dispose(sl->peer, slmsg);
 707                 if (sl->wblocked && sl->repbytes < syslink_bufsize) {
 708                         sl->wblocked = 0;       /* MP race ok here */
 709                         wakeup(&sl->wblocked);
 710                 }
 711         } else {
 712                 /*
 713                  * Leave the command in the RB tree but clear ONINQ now
 714                  * that we have returned it to userland so userland can
 715                  * reply to it.
 716                  */
 717                 slmsg->flags &= ~SLMSGF_ONINQ;
 718         }
 719         return(error);
 720 done1:
 721         spin_unlock_wr(&sl->spin);
 722 done2:
 723         return(error);
 724 }
 725
 726 /*
 727  * Userland writes syslink message (optionally with DMA buffer in iov[1]).
 728  */
 729 static
 730 int
 731 slfileop_write(struct file *fp, struct uio *uio, struct ucred *cred, int flags)
 732 {
 733         struct sldesc *sl = fp->f_data;
 734         struct slmsg *slmsg;
 735         struct slmsg *slcmd;
 736         struct syslink_msg sltmp;
 737         struct syslink_msg *wmsg;       /* wire message */
 738         struct iovec *iov0;
 739         struct iovec *iov1;
 740         sl_proto_t proto;
 741         int nbio;
 742         int error;
 743         int xflags;
 744
 745         /*
 746          * Kinda messy.  Figure out the non-blocking state
 747          */
 748         if (flags & O_FBLOCKING)
 749                 nbio = 0;
 750         else if (flags & O_FNONBLOCKING)
 751                 nbio = 1;
 752         else if (fp->f_flag & O_NONBLOCK)
 753                 nbio = 1;
 754         else
 755                 nbio = 0;
 756
 757         /*
 758          * Validate the uio
 759          */
 760         if (uio->uio_iovcnt < 1) {
 761                 error = 0;
 762                 goto done2;
 763         }
 764         iov0 = &uio->uio_iov[0];
 765         if (iov0->iov_len > SLMSG_BIG) {
 766                 error = EFBIG;
 767                 goto done2;
 768         }
 769         if (uio->uio_iovcnt > 2) {
 770                 error = EFBIG;
 771                 goto done2;
 772         }
 773         if (uio->uio_iovcnt > 1) {
 774                 iov1 = &uio->uio_iov[1];
 775                 if (iov1->iov_len > XIO_INTERNAL_SIZE) {
 776                         error = EFBIG;
 777                         goto done2;
 778                 }
 779                 if ((intptr_t)iov1->iov_base & PAGE_MASK) {
 780                         error = EINVAL;
 781                         goto done2;
 782                 }
 783         } else {
 784                 iov1 = NULL;
 785         }
 786
 787         /*
 788          * Handle the buffer-full case.  slpeer cmdbytes is managed
 789          * by the backend function, not us so if the callback just
 790          * directly implements the message and never adjusts cmdbytes,
 791          * we will never sleep here.
 792          */
 793         if (sl->flags & SLF_WSHUTDOWN) {
 794                 error = EPIPE;
 795                 goto done2;
 796         }
 797
 798         /*
 799          * Only commands can block the pipe, not replies.  Otherwise a
 800          * deadlock is possible.
 801          */
 802         error = copyin(iov0->iov_base, &sltmp, sizeof(sltmp));
 803         if (error)
 804                 goto done2;
 805         if ((proto = sltmp.sm_proto) & SM_PROTO_ENDIAN_REV)
 806                 proto = bswap16(proto);
 807         error = sl->peer->backend_wblocked(sl->peer, nbio, proto);
 808         if (error)
 809                 goto done2;
 810
 811         /*
 812          * Allocate a slmsg and load the message.  Note that the bytes
 813          * returned to userland only reflects the primary syslink message
 814          * and does not include any DMA buffers.
 815          */
 816         if (iov0->iov_len <= SLMSG_SMALL)
 817                 slmsg = objcache_get(sl_objcache_small, M_WAITOK);
 818         else
 819                 slmsg = objcache_get(sl_objcache_big, M_WAITOK);
 820         slmsg->msgsize = iov0->iov_len;
 821         wmsg = slmsg->msg;
 822
 823         error = uiomove((void *)wmsg, iov0->iov_len, uio);
 824         if (error)
 825                 goto done1;
 826         error = syslink_validate_msg(wmsg, slmsg->msgsize);
 827         if (error)
 828                 goto done1;
 829
 830         if ((wmsg->sm_head.se_cmd & SE_CMDF_REPLY) == 0) {
 831                 /*
 832                  * Install the XIO for commands if any DMA flags are set.
 833                  *
 834                  * XIOF_VMLINEAR requires that the XIO represent a
 835                  * contiguous set of pages associated with a single VM
 836                  * object (so the reader side can mmap it easily).
 837                  *
 838                  * XIOF_VMLINEAR might not be set when the kernel sends
 839                  * commands to userland so the reader side backs off to
 840                  * a backup buffer if it isn't set, but we require it
 841                  * for userland writes.
 842                  */
 843                 xflags = XIOF_VMLINEAR;
 844                 if (wmsg->sm_head.se_cmd & SE_CMDF_DMAR)
 845                         xflags |= XIOF_READ | XIOF_WRITE;
 846                 else if (wmsg->sm_head.se_cmd & SE_CMDF_DMAW)
 847                         xflags |= XIOF_READ;
 848                 if (xflags && iov1) {
 849                         get_mplock();
 850                         error = xio_init_ubuf(&slmsg->xio, iov1->iov_base,
 851                                               iov1->iov_len, xflags);
 852                         rel_mplock();
 853                         if (error)
 854                                 goto done1;
 855                         slmsg->flags |= SLMSGF_HASXIO;
 856                 }
 857                 error = sl->peer->backend_write(sl->peer, slmsg);
 858         } else {
 859                 /*
 860                  * Replies have to be matched up against received commands.
 861                  */
 862                 spin_lock_wr(&sl->spin);
 863                 slcmd = slmsg_rb_tree_RB_LOOKUP(&sl->reply_rb_root,
 864                                                 slmsg->msg->sm_msgid);
 865                 if (slcmd == NULL || (slcmd->flags & SLMSGF_ONINQ)) {
 866                         error = ENOENT;
 867                         spin_unlock_wr(&sl->spin);
 868                         goto done1;
 869                 }
 870                 RB_REMOVE(slmsg_rb_tree, &sl->reply_rb_root, slcmd);
 871                 sl->cmdbytes -= slcmd->maxsize;
 872                 spin_unlock_wr(&sl->spin);
 873
 874                 /*
 875                  * If the original command specified DMAR, has an xio, and
 876                  * our write specifies a DMA buffer, then we can do a
 877                  * copyback.  But if we are linearly mapped and the caller
 878                  * is using the map base address, then the caller filled in
 879                  * the data via the direct memory map and no copyback is
 880                  * needed.
 881                  */
 882                 if ((slcmd->msg->sm_head.se_cmd & SE_CMDF_DMAR) && iov1 &&
 883                     (slcmd->flags & SLMSGF_HASXIO) &&
 884                     ((slcmd->flags & SLMSGF_LINMAP) == 0 ||
 885                      iov1->iov_base != slcmd->vmbase)
 886                 ) {
 887                         size_t count;
 888                         if (iov1->iov_len > slcmd->xio.xio_bytes)
 889                                 count = slcmd->xio.xio_bytes;
 890                         else
 891                                 count = iov1->iov_len;
 892                         get_mplock();
 893                         error = xio_copy_utox(&slcmd->xio, 0, iov1->iov_base,
 894                                               count);
 895                         rel_mplock();
 896                 }
 897
 898                 /*
 899                  * If we had mapped a DMA buffer, remove it
 900                  */
 901                 if (slcmd->flags & SLMSGF_LINMAP) {
 902                         get_mplock();
 903                         sl_local_munmap(slcmd);
 904                         rel_mplock();
 905                 }
 906
 907                 /*
 908                  * Reply and handle unblocking
 909                  */
 910                 sl->peer->backend_reply(sl->peer, slcmd, slmsg);
 911                 if (sl->wblocked && sl->cmdbytes < syslink_bufsize) {
 912                         sl->wblocked = 0;       /* MP race ok here */
 913                         wakeup(&sl->wblocked);
 914                 }
 915
 916                 /*
 917                  * slmsg has already been dealt with, make sure error is
 918                  * 0 so we do not double-free it.
 919                  */
 920                 error = 0;
 921         }
 922         /* fall through */
 923 done1:
 924         if (error)
 925                 slmsg_put(slmsg);
 926         /* fall through */
 927 done2:
 928         return(error);
 929 }
 930
 931 /*
 932  * Close a syslink descriptor.
 933  *
 934  * Disassociate the syslink from the file descriptor and disconnect from
 935  * any peer.
 936  */
 937 static
 938 int
 939 slfileop_close(struct file *fp)
 940 {
 941         struct sldesc *sl;
 942
 943         /*
 944          * Disassociate the file pointer.  Take ownership of the ref on the
 945          * sldesc.
 946          */
 947         sl = fp->f_data;
 948         fp->f_data = NULL;
 949         fp->f_ops = &badfileops;
 950         sl->xfp = NULL;
 951
 952         /*
 953          * Shutdown both directions.  The other side will not issue API
 954          * calls to us after we've shutdown both directions.
 955          */
 956         shutdownsldesc(sl, SHUT_RDWR);
 957
 958         /*
 959          * Cleanup
 960          */
 961         KKASSERT(sl->cmdbytes == 0);
 962         KKASSERT(sl->repbytes == 0);
 963         sldrop(sl);
 964         return(0);
 965 }
 966
 967 /*
 968  * MPSAFE
 969  */
 970 static
 971 int
 972 slfileop_stat (struct file *fp, struct stat *sb, struct ucred *cred)
 973 {
 974         return(EINVAL);
 975 }
 976
 977 static
 978 int
 979 slfileop_shutdown (struct file *fp, int how)
 980 {
 981         shutdownsldesc((struct sldesc *)fp->f_data, how);
 982         return(0);
 983 }
 984
 985 static
 986 int
 987 slfileop_ioctl (struct file *fp, u_long cmd, caddr_t data,
 988                 struct ucred *cred, struct sysmsg *msg)
 989 {
 990         return(EINVAL);
 991 }
 992
 993 static
 994 int
 995 slfileop_poll (struct file *fp, int events, struct ucred *cred)
 996 {
 997         return(0);
 998 }
 999
1000 static
1001 int
1002 slfileop_kqfilter(struct file *fp, struct knote *kn)
1003 {
1004         return(0);
1005 }
1006
1007 /************************************************************************
1008  *                          LOCAL MEMORY MAPPING                        *
1009  ************************************************************************
1010  *
1011  * This feature is currently not implemented
1012  *
1013  */
1014
1015 static
1016 int
1017 sl_local_mmap(struct slmsg *slmsg, char *base, size_t len)
1018 {
1019         return (EOPNOTSUPP);
1020 }
1021
1022 static
1023 void
1024 sl_local_munmap(struct slmsg *slmsg)
1025 {
1026         /* empty */
1027 }
1028
1029 #if 0
1030
1031 static
1032 int
1033 sl_local_mmap(struct slmsg *slmsg, char *base, size_t len)
1034 {
1035         struct vmspace *vms = curproc->p_vmspace;
1036         vm_offset_t addr = (vm_offset_t)base;
1037
1038         /* XXX  check user address range */
1039         error = vm_map_replace(
1040                         &vma->vm_map,
1041                         (vm_offset_t)base, (vm_offset_t)base + len,
1042                         slmsg->xio.xio_pages[0]->object,
1043                         slmsg->xio.xio_pages[0]->pindex << PAGE_SHIFT,
1044                         VM_PROT_READ|VM_PROT_WRITE,
1045                         VM_PROT_READ|VM_PROT_WRITE,
1046                         MAP_DISABLE_SYNCER);
1047         }
1048         if (error == 0) {
1049                 slmsg->flags |= SLMSGF_LINMAP;
1050                 slmsg->vmbase = base;
1051                 slmsg->vmsize = len;
1052         }
1053         return (error);
1054 }
1055
1056 static
1057 void
1058 sl_local_munmap(struct slmsg *slmsg)
1059 {
1060         if (slmsg->flags & SLMSGF_LINMAP) {
1061                 vm_map_remove(&curproc->p_vmspace->vm_map,
1062                               slmsg->vmbase,
1063                               slmsg->vmbase + slcmd->vmsize);
1064                 slmsg->flags &= ~SLMSGF_LINMAP;
1065         }
1066 }
1067
1068 #endif
1069
1070 /************************************************************************
1071  *                          MESSAGE VALIDATION                          *
1072  ************************************************************************
1073  *
1074  * Validate that the syslink message.  Check that all headers and elements
1075  * conform.  Correct the endian if necessary.
1076  *
1077  * NOTE: If reverse endian needs to be corrected, SE_CMDF_UNTRANSLATED
1078  * is recursively flipped on all syslink_elm's in the message.  As the
1079  * message traverses the mesh, multiple flips may occur.  It is
1080  * up to the RPC protocol layer to correct opaque data payloads and
1081  * SE_CMDF_UNTRANSLATED prevents the protocol layer from misinterpreting
1082  * a command or reply element which has not been endian-corrected.
1083  */
1084 static
1085 int
1086 syslink_validate_msg(struct syslink_msg *msg, int bytes)
1087 {
1088         int aligned_reclen;
1089         int swapit;
1090         int error;
1091
1092         /*
1093          * The raw message must be properly-aligned.
1094          */
1095         if (bytes & SL_ALIGNMASK)
1096                 return (EINVAL);
1097
1098         while (bytes) {
1099                 /*
1100                  * The message must at least contain the msgid, bytes, and
1101                  * protoid.
1102                  */
1103                 if (bytes < SL_MIN_PAD_SIZE)
1104                         return (EINVAL);
1105
1106                 /*
1107                  * Fix the endian if it is reversed.
1108                  */
1109                 if (msg->sm_proto & SM_PROTO_ENDIAN_REV) {
1110                         msg->sm_msgid = bswap64(msg->sm_msgid);
1111                         msg->sm_sessid = bswap64(msg->sm_sessid);
1112                         msg->sm_bytes = bswap16(msg->sm_bytes);
1113                         msg->sm_proto = bswap16(msg->sm_proto);
1114                         msg->sm_rlabel = bswap32(msg->sm_rlabel);
1115                         if (msg->sm_proto & SM_PROTO_ENDIAN_REV)
1116                                 return (EINVAL);
1117                         swapit = 1;
1118                 } else {
1119                         swapit = 0;
1120                 }
1121
1122                 /*
1123                  * Validate the contents.  For PADs, the entire payload is
1124                  * ignored and the minimum message size can be as small as
1125                  * 8 bytes.
1126                  */
1127                 if (msg->sm_proto == SMPROTO_PAD) {
1128                         if (msg->sm_bytes < SL_MIN_PAD_SIZE ||
1129                             msg->sm_bytes > bytes) {
1130                                 return (EINVAL);
1131                         }
1132                         /* ignore the entire payload, it can be garbage */
1133                 } else {
1134                         if (msg->sm_bytes < SL_MIN_MSG_SIZE ||
1135                             msg->sm_bytes > bytes) {
1136                                 return (EINVAL);
1137                         }
1138                         error = syslink_validate_elm(
1139                                     &msg->sm_head,
1140                                     msg->sm_bytes -
1141                                         offsetof(struct syslink_msg,
1142                                                  sm_head),
1143                                     swapit, SL_MAXDEPTH);
1144                         if (error)
1145                                 return (error);
1146                 }
1147
1148                 /*
1149                  * The aligned payload size must be used to locate the
1150                  * next syslink_msg in the buffer.
1151                  */
1152                 aligned_reclen = SL_MSG_ALIGN(msg->sm_bytes);
1153                 bytes -= aligned_reclen;
1154                 msg = (void *)((char *)msg + aligned_reclen);
1155         }
1156         return(0);
1157 }
1158
1159 static
1160 int
1161 syslink_validate_elm(struct syslink_elm *elm, sl_reclen_t bytes,
1162                      int swapit, int depth)
1163 {
1164         int aligned_reclen;
1165
1166         /*
1167          * If the buffer isn't big enough to fit the header, stop now!
1168          */
1169         if (bytes < SL_MIN_ELM_SIZE)
1170                 return (EINVAL);
1171         /*
1172          * All syslink_elm headers are recursively endian-adjusted.  Opaque
1173          * data payloads are not.
1174          */
1175         if (swapit) {
1176                 elm->se_cmd = bswap16(elm->se_cmd) ^ SE_CMDF_UNTRANSLATED;
1177                 elm->se_bytes = bswap16(elm->se_bytes);
1178                 elm->se_aux = bswap32(elm->se_aux);
1179         }
1180
1181         /*
1182          * Check element size requirements.
1183          */
1184         if (elm->se_bytes < SL_MIN_ELM_SIZE || elm->se_bytes > bytes)
1185                 return (EINVAL);
1186
1187         /*
1188          * Recursively check structured payloads.  A structured payload may
1189          * contain as few as 0 recursive elements.
1190          */
1191         if (elm->se_cmd & SE_CMDF_STRUCTURED) {
1192                 if (depth == 0)
1193                         return (EINVAL);
1194                 bytes -= SL_MIN_ELM_SIZE;
1195                 ++elm;
1196                 while (bytes > 0) {
1197                         if (syslink_validate_elm(elm, bytes, swapit, depth - 1))
1198                                 return (EINVAL);
1199                         aligned_reclen = SL_MSG_ALIGN(elm->se_bytes);
1200                         elm = (void *)((char *)elm + aligned_reclen);
1201                         bytes -= aligned_reclen;
1202                 }
1203         }
1204         return(0);
1205 }
1206
1207 /************************************************************************
1208  *                  BACKEND FUNCTIONS - USER DESCRIPTOR                 *
1209  ************************************************************************
1210  *
1211  * Peer backend links are primarily used when userland creates a pair
1212  * of linked descriptors.
1213  */
1214
1215 /*
1216  * Do any required blocking / nbio handling for attempts to write to
1217  * a sldesc associated with a user descriptor.
1218  */
1219 static
1220 int
1221 backend_wblocked_user(struct sldesc *sl, int nbio, sl_proto_t proto)
1222 {
1223         int error = 0;
1224         int *bytesp = (proto & SM_PROTO_REPLY) ? &sl->repbytes : &sl->cmdbytes;
1225
1226         /*
1227          * Block until sufficient data is drained by the target.  It is
1228          * ok to have a MP race against cmdbytes.
1229          */
1230         if (*bytesp >= syslink_bufsize) {
1231                 spin_lock_wr(&sl->spin);
1232                 while (*bytesp >= syslink_bufsize) {
1233                         if (sl->flags & SLF_WSHUTDOWN) {
1234                                 error = EPIPE;
1235                                 break;
1236                         }
1237                         if (nbio) {
1238                                 error = EAGAIN;
1239                                 break;
1240                         }
1241                         ++sl->wblocked;
1242                         error = ssleep(&sl->wblocked, &sl->spin,
1243                                        PCATCH, "slwmsg", 0);
1244                         if (error)
1245                                 break;
1246                 }
1247                 spin_unlock_wr(&sl->spin);
1248         }
1249         return (error);
1250 }
1251
1252 /*
1253  * Unconditionally write a syslink message to the sldesc associated with
1254  * a user descriptor.  Command messages are also placed in a red-black
1255  * tree so their DMA tag (if any) can be accessed and so they can be
1256  * linked to any reply message.
1257  */
1258 static
1259 int
1260 backend_write_user(struct sldesc *sl, struct slmsg *slmsg)
1261 {
1262         int error;
1263
1264         spin_lock_wr(&sl->spin);
1265         if (sl->flags & SLF_RSHUTDOWN) {
1266                 /*
1267                  * Not accepting new messages
1268                  */
1269                 error = EPIPE;
1270         } else if (slmsg->msg->sm_proto & SM_PROTO_REPLY) {
1271                 /*
1272                  * Write a reply
1273                  */
1274                 TAILQ_INSERT_TAIL(&sl->inq, slmsg, tqnode);
1275                 sl->repbytes += slmsg->maxsize;
1276                 slmsg->flags |= SLMSGF_ONINQ;
1277                 error = 0;
1278         } else if (RB_INSERT(slmsg_rb_tree, &sl->reply_rb_root, slmsg)) {
1279                 /*
1280                  * Write a command, but there was a msgid collision when
1281                  * we tried to insert it into the RB tree.
1282                  */
1283                 error = EEXIST;
1284         } else {
1285                 /*
1286                  * Write a command, successful insertion into the RB tree.
1287                  */
1288                 TAILQ_INSERT_TAIL(&sl->inq, slmsg, tqnode);
1289                 sl->cmdbytes += slmsg->maxsize;
1290                 slmsg->flags |= SLMSGF_ONINQ;
1291                 error = 0;
1292         }
1293         spin_unlock_wr(&sl->spin);
1294         if (sl->rwaiters)
1295                 wakeup(&sl->rwaiters);
1296         return(error);
1297 }
1298
1299 /*
1300  * Our peer is replying a command we previously sent it back to us, along
1301  * with the reply message (if not NULL).  We just queue the reply to
1302  * userland and free of the command.
1303  */
1304 static
1305 void
1306 backend_reply_user(struct sldesc *sl, struct slmsg *slcmd, struct slmsg *slrep)
1307 {
1308         int error;
1309
1310         slmsg_put(slcmd);
1311         if (slrep) {
1312                 spin_lock_wr(&sl->spin);
1313                 if ((sl->flags & SLF_RSHUTDOWN) == 0) {
1314                         TAILQ_INSERT_TAIL(&sl->inq, slrep, tqnode);
1315                         sl->repbytes += slrep->maxsize;
1316                         error = 0;
1317                 } else {
1318                         error = EPIPE;
1319                 }
1320                 spin_unlock_wr(&sl->spin);
1321                 if (error)
1322                         sl->peer->backend_dispose(sl->peer, slrep);
1323                 else if (sl->rwaiters)
1324                         wakeup(&sl->rwaiters);
1325         }
1326 }
1327
1328 static
1329 void
1330 backend_dispose_user(struct sldesc *sl, struct slmsg *slmsg)
1331 {
1332         slmsg_put(slmsg);
1333 }
1334
1335 /************************************************************************
1336  *                      KERNEL DRIVER OR FILESYSTEM API                 *
1337  ************************************************************************
1338  *
1339  */
1340
1341 /*
1342  * Create a user<->kernel link, returning the user descriptor in *fdp
1343  * and the kernel descriptor in *kslp.  0 is returned on success, and an
1344  * error code is returned on failure.
1345  */
1346 int
1347 syslink_ukbackend(int *pfd, struct sldesc **kslp)
1348 {
1349         struct thread *td = curthread;
1350         struct filedesc *fdp = td->td_proc->p_fd;
1351         struct file *fp;
1352         struct sldesc *usl;
1353         struct sldesc *ksl;
1354         int error;
1355         int fd;
1356
1357         *pfd = -1;
1358         *kslp = NULL;
1359
1360         error = falloc(td->td_lwp, &fp, &fd);
1361         if (error)
1362                 return(error);
1363         usl = allocsldesc(NULL);
1364         usl->backend_wblocked = backend_wblocked_user;
1365         usl->backend_write = backend_write_user;
1366         usl->backend_reply = backend_reply_user;
1367         usl->backend_dispose = backend_dispose_user;
1368
1369         ksl = allocsldesc(usl->common);
1370         ksl->peer = usl;
1371         ksl->backend_wblocked = backend_wblocked_kern;
1372         ksl->backend_write = backend_write_kern;
1373         ksl->backend_reply = backend_reply_kern;
1374         ksl->backend_dispose = backend_dispose_kern;
1375
1376         usl->peer = ksl;
1377
1378         setsldescfp(usl, fp);
1379         fsetfd(fdp, fp, fd);
1380         fdrop(fp);
1381
1382         *pfd = fd;
1383         *kslp = ksl;
1384         return(0);
1385 }
1386
1387 /*
1388  * Assign a unique message id, issue a syslink message to userland,
1389  * and wait for a reply.
1390  */
1391 int
1392 syslink_kdomsg(struct sldesc *ksl, struct slmsg *slmsg)
1393 {
1394         struct syslink_msg *msg;
1395         int error;
1396
1397         /*
1398          * Finish initializing slmsg and post it to the red-black tree for
1399          * reply matching.  If the message id is already in use we return
1400          * EEXIST, giving the originator the chance to roll a new msgid.
1401          */
1402         msg = slmsg->msg;
1403         slmsg->msgsize = msg->sm_bytes;
1404         if ((error = syslink_validate_msg(msg, msg->sm_bytes)) != 0)
1405                 return (error);
1406         msg->sm_msgid = allocsysid();
1407
1408         /*
1409          * Issue the request and wait for a matching reply or failure,
1410          * then remove the message from the matching tree and return.
1411          */
1412         error = ksl->peer->backend_write(ksl->peer, slmsg);
1413         spin_lock_wr(&ksl->spin);
1414         if (error == 0) {
1415                 while (slmsg->rep == NULL) {
1416                         error = ssleep(slmsg, &ksl->spin, 0, "kwtmsg", 0);
1417                         /* XXX ignore error for now */
1418                 }
1419                 if (slmsg->rep == (struct slmsg *)-1) {
1420                         error = EIO;
1421                         slmsg->rep = NULL;
1422                 } else {
1423                         error = slmsg->rep->msg->sm_head.se_aux;
1424                 }
1425         }
1426         spin_unlock_wr(&ksl->spin);
1427         return(error);
1428 }
1429
1430 /*
1431  * Similar to syslink_kdomsg but return immediately instead of
1432  * waiting for a reply.  The kernel must supply a callback function
1433  * which will be made in the context of the user process replying
1434  * to the message.
1435  */
1436 int
1437 syslink_ksendmsg(struct sldesc *ksl, struct slmsg *slmsg,
1438                  void (*func)(struct slmsg *, void *, int), void *arg)
1439 {
1440         struct syslink_msg *msg;
1441         int error;
1442
1443         /*
1444          * Finish initializing slmsg and post it to the red-black tree for
1445          * reply matching.  If the message id is already in use we return
1446          * EEXIST, giving the originator the chance to roll a new msgid.
1447          */
1448         msg = slmsg->msg;
1449         slmsg->msgsize = msg->sm_bytes;
1450         slmsg->callback_func = func;
1451         slmsg->callback_data = arg;
1452         if ((error = syslink_validate_msg(msg, msg->sm_bytes)) != 0)
1453                 return (error);
1454         msg->sm_msgid = allocsysid();
1455
1456         /*
1457          * Issue the request.  If no error occured the operation will be
1458          * in progress, otherwise the operation is considered to have failed
1459          * and the caller can deallocate the slmsg.
1460          */
1461         error = ksl->peer->backend_write(ksl->peer, slmsg);
1462         return (error);
1463 }
1464
1465 int
1466 syslink_kwaitmsg(struct sldesc *ksl, struct slmsg *slmsg)
1467 {
1468         int error;
1469
1470         spin_lock_wr(&ksl->spin);
1471         while (slmsg->rep == NULL) {
1472                 error = ssleep(slmsg, &ksl->spin, 0, "kwtmsg", 0);
1473                 /* XXX ignore error for now */
1474         }
1475         if (slmsg->rep == (struct slmsg *)-1) {
1476                 error = EIO;
1477                 slmsg->rep = NULL;
1478         } else {
1479                 error = slmsg->rep->msg->sm_head.se_aux;
1480         }
1481         spin_unlock_wr(&ksl->spin);
1482         return(error);
1483 }
1484
1485 struct slmsg *
1486 syslink_kallocmsg(void)
1487 {
1488         return(objcache_get(sl_objcache_small, M_WAITOK));
1489 }
1490
1491 void
1492 syslink_kfreemsg(struct sldesc *ksl, struct slmsg *slmsg)
1493 {
1494         struct slmsg *rep;
1495
1496         if ((rep = slmsg->rep) != NULL) {
1497                 slmsg->rep = NULL;
1498                 ksl->peer->backend_dispose(ksl->peer, rep);
1499         }
1500         slmsg->callback_func = NULL;
1501         slmsg_put(slmsg);
1502 }
1503
1504 void
1505 syslink_kshutdown(struct sldesc *ksl, int how)
1506 {
1507         shutdownsldesc(ksl, how);
1508 }
1509
1510 void
1511 syslink_kclose(struct sldesc *ksl)
1512 {
1513         shutdownsldesc(ksl, SHUT_RDWR);
1514         sldrop(ksl);
1515 }
1516
1517 /*
1518  * Associate a DMA buffer with a kernel syslink message prior to it
1519  * being sent to userland.  The DMA buffer is set up from the point
1520  * of view of the target.
1521  */
1522 int
1523 syslink_kdmabuf_pages(struct slmsg *slmsg, struct vm_page **mbase, int npages)
1524 {
1525         int xflags;
1526         int error;
1527
1528         xflags = XIOF_VMLINEAR;
1529         if (slmsg->msg->sm_head.se_cmd & SE_CMDF_DMAR)
1530                 xflags |= XIOF_READ | XIOF_WRITE;
1531         else if (slmsg->msg->sm_head.se_cmd & SE_CMDF_DMAW)
1532                 xflags |= XIOF_READ;
1533         error = xio_init_pages(&slmsg->xio, mbase, npages, xflags);
1534         slmsg->flags |= SLMSGF_HASXIO;
1535         return (error);
1536 }
1537
1538 /*
1539  * Associate a DMA buffer with a kernel syslink message prior to it
1540  * being sent to userland.  The DMA buffer is set up from the point
1541  * of view of the target.
1542  */
1543 int
1544 syslink_kdmabuf_data(struct slmsg *slmsg, char *base, int bytes)
1545 {
1546         int xflags;
1547
1548         xflags = XIOF_VMLINEAR;
1549         if (slmsg->msg->sm_head.se_cmd & SE_CMDF_DMAR)
1550                 xflags |= XIOF_READ | XIOF_WRITE;
1551         else if (slmsg->msg->sm_head.se_cmd & SE_CMDF_DMAW)
1552                 xflags |= XIOF_READ;
1553         xio_init_kbuf(&slmsg->xio, base, bytes);
1554         slmsg->xio.xio_flags |= xflags;
1555         slmsg->flags |= SLMSGF_HASXIO;
1556         return(0);
1557 }
1558
1559 /************************************************************************
1560  *                  BACKEND FUNCTIONS FOR KERNEL API                    *
1561  ************************************************************************
1562  *
1563  * These are the backend functions for a sldesc associated with a kernel
1564  * API.
1565  */
1566
1567 /*
1568  * Our peer wants to write a syslink message to us and is asking us to
1569  * block if our input queue is full.  We don't implement command reception
1570  * so don't block right now.
1571  */
1572 static
1573 int
1574 backend_wblocked_kern(struct sldesc *ksl, int nbio, sl_proto_t proto)
1575 {
1576         /* never blocks */
1577         return(0);
1578 }
1579
1580 /*
1581  * Our peer is writing a request to the kernel.  At the moment we do not
1582  * accept commands.
1583  */
1584 static
1585 int
1586 backend_write_kern(struct sldesc *ksl, struct slmsg *slmsg)
1587 {
1588         return(EOPNOTSUPP);
1589 }
1590
1591 /*
1592  * Our peer wants to reply to a syslink message we sent it earlier.  The
1593  * original command (that we passed to our peer), and the peer's reply
1594  * is specified.  If the peer has failed slrep will be NULL.
1595  */
1596 static
1597 void
1598 backend_reply_kern(struct sldesc *ksl, struct slmsg *slcmd, struct slmsg *slrep)
1599 {
1600         int error;
1601
1602         spin_lock_wr(&ksl->spin);
1603         if (slrep == NULL) {
1604                 slcmd->rep = (struct slmsg *)-1;
1605                 error = EIO;
1606         } else {
1607                 slcmd->rep = slrep;
1608                 error = slrep->msg->sm_head.se_aux;
1609         }
1610         spin_unlock_wr(&ksl->spin);
1611
1612         /*
1613          * Issue callback or wakeup a synchronous waiter.
1614          */
1615         if (slcmd->callback_func) {
1616                 slcmd->callback_func(slcmd, slcmd->callback_data, error);
1617         } else {
1618                 wakeup(slcmd);
1619         }
1620 }
1621
1622 /*
1623  * Any reply messages we sent to our peer are returned to us for disposal.
1624  * Since we do not currently accept commands from our peer, there will not
1625  * be any replies returned to the peer to dispose of.
1626  */
1627 static
1628 void
1629 backend_dispose_kern(struct sldesc *ksl, struct slmsg *slmsg)
1630 {
1631         panic("backend_dispose_kern: kernel can't accept commands so it "
1632               "certainly did not reply to one!");
1633 }
1634