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