sys/kern/lwkt_msgport.c

   1 /*
   2  * Copyright (c) 2003,2004 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  * NOTE! This file may be compiled for userland libraries as well as for
  35  * the kernel.
  36  */
  37
  38 #include <sys/param.h>
  39 #include <sys/systm.h>
  40 #include <sys/kernel.h>
  41 #include <sys/proc.h>
  42 #include <sys/rtprio.h>
  43 #include <sys/queue.h>
  44 #include <sys/sysctl.h>
  45 #include <sys/kthread.h>
  46 #include <sys/signalvar.h>
  47 #include <sys/signal2.h>
  48 #include <machine/cpu.h>
  49 #include <sys/lock.h>
  50
  51 #include <vm/vm.h>
  52 #include <vm/vm_param.h>
  53 #include <vm/vm_kern.h>
  54 #include <vm/vm_object.h>
  55 #include <vm/vm_page.h>
  56 #include <vm/vm_map.h>
  57 #include <vm/vm_pager.h>
  58 #include <vm/vm_extern.h>
  59 #include <vm/vm_zone.h>
  60
  61 #include <sys/thread2.h>
  62 #include <sys/msgport2.h>
  63 #include <sys/spinlock2.h>
  64 #include <sys/serialize.h>
  65
  66 #include <machine/stdarg.h>
  67 #include <machine/cpufunc.h>
  68 #ifdef SMP
  69 #include <machine/smp.h>
  70 #endif
  71
  72 #include <sys/malloc.h>
  73 MALLOC_DEFINE(M_LWKTMSG, "lwkt message", "lwkt message");
  74
  75 /************************************************************************
  76  *                              MESSAGE FUNCTIONS                       *
  77  ************************************************************************/
  78
  79 /*
  80  * lwkt_sendmsg()
  81  *
  82  *      Request asynchronous completion and call lwkt_beginmsg().  The
  83  *      target port can opt to execute the message synchronously or
  84  *      asynchronously and this function will automatically queue the
  85  *      response if the target executes the message synchronously.
  86  *
  87  *      NOTE: The message is in an indeterminant state until this call
  88  *      returns.  The caller should not mess with it (e.g. try to abort it)
  89  *      until then.
  90  */
  91 void
  92 lwkt_sendmsg(lwkt_port_t port, lwkt_msg_t msg)
  93 {
  94     int error;
  95
  96     KKASSERT(msg->ms_reply_port != NULL &&
  97              (msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == MSGF_DONE);
  98     msg->ms_flags &= ~(MSGF_REPLY | MSGF_SYNC | MSGF_DONE);
  99     if ((error = lwkt_beginmsg(port, msg)) != EASYNC) {
 100         lwkt_replymsg(msg, error);
 101     }
 102 }
 103
 104 /*
 105  * lwkt_domsg()
 106  *
 107  *      Request asynchronous completion and call lwkt_beginmsg().  The
 108  *      target port can opt to execute the message synchronously or
 109  *      asynchronously and this function will automatically queue the
 110  *      response if the target executes the message synchronously.
 111  */
 112 int
 113 lwkt_domsg(lwkt_port_t port, lwkt_msg_t msg, int flags)
 114 {
 115     int error;
 116
 117     KKASSERT(msg->ms_reply_port != NULL &&
 118              (msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == MSGF_DONE);
 119     msg->ms_flags &= ~(MSGF_REPLY | MSGF_DONE);
 120     msg->ms_flags |= MSGF_SYNC;
 121     if ((error = lwkt_beginmsg(port, msg)) == EASYNC) {
 122         error = lwkt_waitmsg(msg, flags);
 123     } else {
 124         msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
 125     }
 126     return(error);
 127 }
 128
 129 /*
 130  * lwkt_forwardmsg()
 131  *
 132  * Forward a message received on one port to another port.
 133  */
 134 int
 135 lwkt_forwardmsg(lwkt_port_t port, lwkt_msg_t msg)
 136 {
 137     int error;
 138
 139     crit_enter();
 140     KKASSERT((msg->ms_flags & (MSGF_QUEUED|MSGF_DONE|MSGF_REPLY)) == 0);
 141     if ((error = port->mp_putport(port, msg)) != EASYNC)
 142         lwkt_replymsg(msg, error);
 143     crit_exit();
 144     return(error);
 145 }
 146
 147 /*
 148  * lwkt_abortmsg()
 149  *
 150  * Attempt to abort a message.  This only works if MSGF_ABORTABLE is set.
 151  * The caller must ensure that the message will not be both replied AND
 152  * destroyed while the abort is in progress.
 153  *
 154  * This function issues a callback which might block!
 155  */
 156 void
 157 lwkt_abortmsg(lwkt_msg_t msg)
 158 {
 159     /*
 160      * A critical section protects us from reply IPIs on this cpu.
 161      */
 162     crit_enter();
 163
 164     /*
 165      * Shortcut the operation if the message has already been returned.
 166      * The callback typically constructs a lwkt_msg with the abort request,
 167      * issues it synchronously, and waits for completion.  The callback
 168      * is not required to actually abort the message and the target port,
 169      * upon receiving an abort request message generated by the callback
 170      * should check whether the original message has already completed or
 171      * not.
 172      */
 173     if (msg->ms_flags & MSGF_ABORTABLE) {
 174         if ((msg->ms_flags & (MSGF_DONE|MSGF_REPLY)) == 0)
 175             msg->ms_abortfn(msg);
 176     }
 177     crit_exit();
 178 }
 179
 180 /************************************************************************
 181  *                      PORT INITIALIZATION API                         *
 182  ************************************************************************/
 183
 184 static void *lwkt_thread_getport(lwkt_port_t port);
 185 static int lwkt_thread_putport(lwkt_port_t port, lwkt_msg_t msg);
 186 static int lwkt_thread_waitmsg(lwkt_msg_t msg, int flags);
 187 static void *lwkt_thread_waitport(lwkt_port_t port, int flags);
 188 static void lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg);
 189 static void lwkt_thread_dropmsg(lwkt_port_t port, lwkt_msg_t msg);
 190
 191 static void *lwkt_spin_getport(lwkt_port_t port);
 192 static int lwkt_spin_putport(lwkt_port_t port, lwkt_msg_t msg);
 193 static int lwkt_spin_waitmsg(lwkt_msg_t msg, int flags);
 194 static void *lwkt_spin_waitport(lwkt_port_t port, int flags);
 195 static void lwkt_spin_replyport(lwkt_port_t port, lwkt_msg_t msg);
 196 static void lwkt_spin_dropmsg(lwkt_port_t port, lwkt_msg_t msg);
 197
 198 static void *lwkt_serialize_getport(lwkt_port_t port);
 199 static int lwkt_serialize_putport(lwkt_port_t port, lwkt_msg_t msg);
 200 static int lwkt_serialize_waitmsg(lwkt_msg_t msg, int flags);
 201 static void *lwkt_serialize_waitport(lwkt_port_t port, int flags);
 202 static void lwkt_serialize_replyport(lwkt_port_t port, lwkt_msg_t msg);
 203
 204 static void lwkt_null_replyport(lwkt_port_t port, lwkt_msg_t msg);
 205 static void *lwkt_panic_getport(lwkt_port_t port);
 206 static int lwkt_panic_putport(lwkt_port_t port, lwkt_msg_t msg);
 207 static int lwkt_panic_waitmsg(lwkt_msg_t msg, int flags);
 208 static void *lwkt_panic_waitport(lwkt_port_t port, int flags);
 209 static void lwkt_panic_replyport(lwkt_port_t port, lwkt_msg_t msg);
 210 static void lwkt_panic_dropmsg(lwkt_port_t port, lwkt_msg_t msg);
 211
 212 /*
 213  * Core port initialization (internal)
 214  */
 215 static __inline
 216 void
 217 _lwkt_initport(lwkt_port_t port,
 218                void *(*gportfn)(lwkt_port_t),
 219                int (*pportfn)(lwkt_port_t, lwkt_msg_t),
 220                int (*wmsgfn)(lwkt_msg_t, int),
 221                void *(*wportfn)(lwkt_port_t, int),
 222                void (*rportfn)(lwkt_port_t, lwkt_msg_t),
 223                void (*dmsgfn)(lwkt_port_t, lwkt_msg_t))
 224 {
 225     bzero(port, sizeof(*port));
 226     TAILQ_INIT(&port->mp_msgq);
 227     TAILQ_INIT(&port->mp_msgq_prio);
 228     port->mp_getport = gportfn;
 229     port->mp_putport = pportfn;
 230     port->mp_waitmsg =  wmsgfn;
 231     port->mp_waitport =  wportfn;
 232     port->mp_replyport = rportfn;
 233     port->mp_dropmsg = dmsgfn;
 234 }
 235
 236 /*
 237  * Schedule the target thread.  If the message flags contains MSGF_NORESCHED
 238  * we tell the scheduler not to reschedule if td is at a higher priority.
 239  *
 240  * This routine is called even if the thread is already scheduled.
 241  */
 242 static __inline
 243 void
 244 _lwkt_schedule_msg(thread_t td, int flags)
 245 {
 246     lwkt_schedule(td);
 247 }
 248
 249 /*
 250  * lwkt_initport_thread()
 251  *
 252  *      Initialize a port for use by a particular thread.  The port may
 253  *      only be used by <td>.
 254  */
 255 void
 256 lwkt_initport_thread(lwkt_port_t port, thread_t td)
 257 {
 258     _lwkt_initport(port,
 259                    lwkt_thread_getport,
 260                    lwkt_thread_putport,
 261                    lwkt_thread_waitmsg,
 262                    lwkt_thread_waitport,
 263                    lwkt_thread_replyport,
 264                    lwkt_thread_dropmsg);
 265     port->mpu_td = td;
 266 }
 267
 268 /*
 269  * lwkt_initport_spin()
 270  *
 271  *      Initialize a port for use with descriptors that might be accessed
 272  *      via multiple LWPs, processes, or threads.  Has somewhat more
 273  *      overhead then thread ports.
 274  */
 275 void
 276 lwkt_initport_spin(lwkt_port_t port)
 277 {
 278     _lwkt_initport(port,
 279                    lwkt_spin_getport,
 280                    lwkt_spin_putport,
 281                    lwkt_spin_waitmsg,
 282                    lwkt_spin_waitport,
 283                    lwkt_spin_replyport,
 284                    lwkt_spin_dropmsg);
 285     spin_init(&port->mpu_spin);
 286 }
 287
 288 /*
 289  * lwkt_initport_serialize()
 290  *
 291  *      Initialize a port for use with descriptors that might be accessed
 292  *      via multiple LWPs, processes, or threads.  Callers are assumed to
 293  *      have held the serializer (slz).
 294  */
 295 void
 296 lwkt_initport_serialize(lwkt_port_t port, struct lwkt_serialize *slz)
 297 {
 298     _lwkt_initport(port,
 299                    lwkt_serialize_getport,
 300                    lwkt_serialize_putport,
 301                    lwkt_serialize_waitmsg,
 302                    lwkt_serialize_waitport,
 303                    lwkt_serialize_replyport,
 304                    lwkt_panic_dropmsg);
 305     port->mpu_serialize = slz;
 306 }
 307
 308 /*
 309  * Similar to the standard initport, this function simply marks the message
 310  * as being done and does not attempt to return it to an originating port.
 311  */
 312 void
 313 lwkt_initport_replyonly_null(lwkt_port_t port)
 314 {
 315     _lwkt_initport(port,
 316                    lwkt_panic_getport,
 317                    lwkt_panic_putport,
 318                    lwkt_panic_waitmsg,
 319                    lwkt_panic_waitport,
 320                    lwkt_null_replyport,
 321                    lwkt_panic_dropmsg);
 322 }
 323
 324 /*
 325  * Initialize a reply-only port, typically used as a message sink.  Such
 326  * ports can only be used as a reply port.
 327  */
 328 void
 329 lwkt_initport_replyonly(lwkt_port_t port,
 330                         void (*rportfn)(lwkt_port_t, lwkt_msg_t))
 331 {
 332     _lwkt_initport(port, lwkt_panic_getport, lwkt_panic_putport,
 333                          lwkt_panic_waitmsg, lwkt_panic_waitport,
 334                          rportfn, lwkt_panic_dropmsg);
 335 }
 336
 337 void
 338 lwkt_initport_putonly(lwkt_port_t port,
 339                       int (*pportfn)(lwkt_port_t, lwkt_msg_t))
 340 {
 341     _lwkt_initport(port, lwkt_panic_getport, pportfn,
 342                          lwkt_panic_waitmsg, lwkt_panic_waitport,
 343                          lwkt_panic_replyport, lwkt_panic_dropmsg);
 344 }
 345
 346 void
 347 lwkt_initport_panic(lwkt_port_t port)
 348 {
 349     _lwkt_initport(port,
 350                    lwkt_panic_getport, lwkt_panic_putport,
 351                    lwkt_panic_waitmsg, lwkt_panic_waitport,
 352                    lwkt_panic_replyport, lwkt_panic_dropmsg);
 353 }
 354
 355 static __inline
 356 void
 357 _lwkt_pullmsg(lwkt_port_t port, lwkt_msg_t msg)
 358 {
 359     lwkt_msg_queue *queue;
 360
 361     /*
 362      * normal case, remove and return the message.
 363      */
 364     if (__predict_false(msg->ms_flags & MSGF_PRIORITY))
 365         queue = &port->mp_msgq_prio;
 366     else
 367         queue = &port->mp_msgq;
 368     TAILQ_REMOVE(queue, msg, ms_node);
 369
 370     /*
 371      * atomic op needed for spin ports
 372      */
 373     atomic_clear_int(&msg->ms_flags, MSGF_QUEUED);
 374 }
 375
 376 static __inline
 377 void
 378 _lwkt_pushmsg(lwkt_port_t port, lwkt_msg_t msg)
 379 {
 380     lwkt_msg_queue *queue;
 381
 382     /*
 383      * atomic op needed for spin ports
 384      */
 385     atomic_set_int(&msg->ms_flags, MSGF_QUEUED);
 386     if (__predict_false(msg->ms_flags & MSGF_PRIORITY))
 387         queue = &port->mp_msgq_prio;
 388     else
 389         queue = &port->mp_msgq;
 390     TAILQ_INSERT_TAIL(queue, msg, ms_node);
 391 }
 392
 393 static __inline
 394 lwkt_msg_t
 395 _lwkt_pollmsg(lwkt_port_t port)
 396 {
 397     lwkt_msg_t msg;
 398
 399     msg = TAILQ_FIRST(&port->mp_msgq_prio);
 400     if (__predict_false(msg != NULL))
 401         return msg;
 402
 403     /*
 404      * Priority queue has no message, fallback to non-priority queue.
 405      */
 406     return TAILQ_FIRST(&port->mp_msgq);
 407 }
 408
 409 static __inline
 410 void
 411 _lwkt_enqueue_reply(lwkt_port_t port, lwkt_msg_t msg)
 412 {
 413     /*
 414      * atomic op needed for spin ports
 415      */
 416     _lwkt_pushmsg(port, msg);
 417     atomic_set_int(&msg->ms_flags, MSGF_REPLY | MSGF_DONE);
 418 }
 419
 420 /************************************************************************
 421  *                      THREAD PORT BACKEND                             *
 422  ************************************************************************
 423  *
 424  * This backend is used when the port a message is retrieved from is owned
 425  * by a single thread (the calling thread).  Messages are IPId to the
 426  * correct cpu before being enqueued to a port.  Note that this is fairly
 427  * optimal since scheduling would have had to do an IPI anyway if the
 428  * message were headed to a different cpu.
 429  */
 430
 431 #ifdef SMP
 432
 433 /*
 434  * This function completes reply processing for the default case in the
 435  * context of the originating cpu.
 436  */
 437 static
 438 void
 439 lwkt_thread_replyport_remote(lwkt_msg_t msg)
 440 {
 441     lwkt_port_t port = msg->ms_reply_port;
 442     int flags;
 443
 444     /*
 445      * Chase any thread migration that occurs
 446      */
 447     if (port->mpu_td->td_gd != mycpu) {
 448         lwkt_send_ipiq(port->mpu_td->td_gd,
 449                        (ipifunc1_t)lwkt_thread_replyport_remote, msg);
 450         return;
 451     }
 452
 453     /*
 454      * Cleanup
 455      */
 456 #ifdef INVARIANTS
 457     KKASSERT(msg->ms_flags & MSGF_INTRANSIT);
 458     msg->ms_flags &= ~MSGF_INTRANSIT;
 459 #endif
 460     flags = msg->ms_flags;
 461     if (msg->ms_flags & MSGF_SYNC) {
 462         cpu_sfence();
 463         msg->ms_flags |= MSGF_REPLY | MSGF_DONE;
 464     } else {
 465         _lwkt_enqueue_reply(port, msg);
 466     }
 467     if (port->mp_flags & MSGPORTF_WAITING)
 468         _lwkt_schedule_msg(port->mpu_td, flags);
 469 }
 470
 471 #endif
 472
 473 /*
 474  * lwkt_thread_replyport() - Backend to lwkt_replymsg()
 475  *
 476  * Called with the reply port as an argument but in the context of the
 477  * original target port.  Completion must occur on the target port's
 478  * cpu.
 479  *
 480  * The critical section protects us from IPIs on the this CPU.
 481  */
 482 static
 483 void
 484 lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg)
 485 {
 486     int flags;
 487
 488     KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED|MSGF_INTRANSIT)) == 0);
 489
 490     if (msg->ms_flags & MSGF_SYNC) {
 491         /*
 492          * If a synchronous completion has been requested, just wakeup
 493          * the message without bothering to queue it to the target port.
 494          *
 495          * Assume the target thread is non-preemptive, so no critical
 496          * section is required.
 497          */
 498 #ifdef SMP
 499         if (port->mpu_td->td_gd == mycpu) {
 500 #endif
 501             flags = msg->ms_flags;
 502             cpu_sfence();
 503             msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
 504             if (port->mp_flags & MSGPORTF_WAITING)
 505                 _lwkt_schedule_msg(port->mpu_td, flags);
 506 #ifdef SMP
 507         } else {
 508 #ifdef INVARIANTS
 509             msg->ms_flags |= MSGF_INTRANSIT;
 510 #endif
 511             msg->ms_flags |= MSGF_REPLY;
 512             lwkt_send_ipiq(port->mpu_td->td_gd,
 513                            (ipifunc1_t)lwkt_thread_replyport_remote, msg);
 514         }
 515 #endif
 516     } else {
 517         /*
 518          * If an asynchronous completion has been requested the message
 519          * must be queued to the reply port.
 520          *
 521          * A critical section is required to interlock the port queue.
 522          */
 523 #ifdef SMP
 524         if (port->mpu_td->td_gd == mycpu) {
 525 #endif
 526             crit_enter();
 527             _lwkt_enqueue_reply(port, msg);
 528             if (port->mp_flags & MSGPORTF_WAITING)
 529                 _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
 530             crit_exit();
 531 #ifdef SMP
 532         } else {
 533 #ifdef INVARIANTS
 534             msg->ms_flags |= MSGF_INTRANSIT;
 535 #endif
 536             msg->ms_flags |= MSGF_REPLY;
 537             lwkt_send_ipiq(port->mpu_td->td_gd,
 538                            (ipifunc1_t)lwkt_thread_replyport_remote, msg);
 539         }
 540 #endif
 541     }
 542 }
 543
 544 /*
 545  * lwkt_thread_dropmsg() - Backend to lwkt_dropmsg()
 546  *
 547  * This function could _only_ be used when caller is in the same thread
 548  * as the message's target port owner thread.
 549  */
 550 static void
 551 lwkt_thread_dropmsg(lwkt_port_t port, lwkt_msg_t msg)
 552 {
 553     KASSERT(port->mpu_td == curthread,
 554             ("message could only be dropped in the same thread "
 555              "as the message target port thread"));
 556     crit_enter_quick(port->mpu_td);
 557     _lwkt_pullmsg(port, msg);
 558     msg->ms_flags |= MSGF_DONE;
 559     crit_exit_quick(port->mpu_td);
 560 }
 561
 562 /*
 563  * lwkt_thread_putport() - Backend to lwkt_beginmsg()
 564  *
 565  * Called with the target port as an argument but in the context of the
 566  * reply port.  This function always implements an asynchronous put to
 567  * the target message port, and thus returns EASYNC.
 568  *
 569  * The message must already have cleared MSGF_DONE and MSGF_REPLY
 570  */
 571
 572 #ifdef SMP
 573
 574 static
 575 void
 576 lwkt_thread_putport_remote(lwkt_msg_t msg)
 577 {
 578     lwkt_port_t port = msg->ms_target_port;
 579
 580     /*
 581      * Chase any thread migration that occurs
 582      */
 583     if (port->mpu_td->td_gd != mycpu) {
 584         lwkt_send_ipiq(port->mpu_td->td_gd,
 585                        (ipifunc1_t)lwkt_thread_putport_remote, msg);
 586         return;
 587     }
 588
 589     /*
 590      * Cleanup
 591      */
 592 #ifdef INVARIANTS
 593     KKASSERT(msg->ms_flags & MSGF_INTRANSIT);
 594     msg->ms_flags &= ~MSGF_INTRANSIT;
 595 #endif
 596     _lwkt_pushmsg(port, msg);
 597     if (port->mp_flags & MSGPORTF_WAITING)
 598         _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
 599 }
 600
 601 #endif
 602
 603 static
 604 int
 605 lwkt_thread_putport(lwkt_port_t port, lwkt_msg_t msg)
 606 {
 607     KKASSERT((msg->ms_flags & (MSGF_DONE | MSGF_REPLY)) == 0);
 608
 609     msg->ms_target_port = port;
 610 #ifdef SMP
 611     if (port->mpu_td->td_gd == mycpu) {
 612 #endif
 613         crit_enter();
 614         _lwkt_pushmsg(port, msg);
 615         if (port->mp_flags & MSGPORTF_WAITING)
 616             _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
 617         crit_exit();
 618 #ifdef SMP
 619     } else {
 620 #ifdef INVARIANTS
 621         msg->ms_flags |= MSGF_INTRANSIT;
 622 #endif
 623         lwkt_send_ipiq(port->mpu_td->td_gd,
 624                         (ipifunc1_t)lwkt_thread_putport_remote, msg);
 625     }
 626 #endif
 627     return (EASYNC);
 628 }
 629
 630 /*
 631  * lwkt_thread_getport()
 632  *
 633  *      Retrieve the next message from the port or NULL if no messages
 634  *      are ready.
 635  */
 636 static
 637 void *
 638 lwkt_thread_getport(lwkt_port_t port)
 639 {
 640     lwkt_msg_t msg;
 641
 642     KKASSERT(port->mpu_td == curthread);
 643
 644     crit_enter_quick(port->mpu_td);
 645     if ((msg = _lwkt_pollmsg(port)) != NULL)
 646         _lwkt_pullmsg(port, msg);
 647     crit_exit_quick(port->mpu_td);
 648     return(msg);
 649 }
 650
 651 /*
 652  * lwkt_thread_waitmsg()
 653  *
 654  *      Wait for a particular message to be replied.  We must be the only
 655  *      thread waiting on the message.  The port must be owned by the
 656  *      caller.
 657  */
 658 static
 659 int
 660 lwkt_thread_waitmsg(lwkt_msg_t msg, int flags)
 661 {
 662     KASSERT((msg->ms_flags & MSGF_DROPABLE) == 0,
 663             ("can't wait dropable message"));
 664
 665     if ((msg->ms_flags & MSGF_DONE) == 0) {
 666         /*
 667          * If the done bit was not set we have to block until it is.
 668          */
 669         lwkt_port_t port = msg->ms_reply_port;
 670         thread_t td = curthread;
 671         int sentabort;
 672
 673         KKASSERT(port->mpu_td == td);
 674         crit_enter_quick(td);
 675         sentabort = 0;
 676
 677         while ((msg->ms_flags & MSGF_DONE) == 0) {
 678             port->mp_flags |= MSGPORTF_WAITING;
 679             if (sentabort == 0) {
 680                 if ((sentabort = lwkt_sleep("waitmsg", flags)) != 0) {
 681                     lwkt_abortmsg(msg);
 682                 }
 683             } else {
 684                 lwkt_sleep("waitabt", 0);
 685             }
 686             port->mp_flags &= ~MSGPORTF_WAITING;
 687         }
 688         if (msg->ms_flags & MSGF_QUEUED)
 689             _lwkt_pullmsg(port, msg);
 690         crit_exit_quick(td);
 691     } else {
 692         /*
 693          * If the done bit was set we only have to mess around with the
 694          * message if it is queued on the reply port.
 695          */
 696         if (msg->ms_flags & MSGF_QUEUED) {
 697             lwkt_port_t port = msg->ms_reply_port;
 698             thread_t td = curthread;
 699
 700             KKASSERT(port->mpu_td == td);
 701             crit_enter_quick(td);
 702             _lwkt_pullmsg(port, msg);
 703             crit_exit_quick(td);
 704         }
 705     }
 706     return(msg->ms_error);
 707 }
 708
 709 static
 710 void *
 711 lwkt_thread_waitport(lwkt_port_t port, int flags)
 712 {
 713     thread_t td = curthread;
 714     lwkt_msg_t msg;
 715     int error;
 716
 717     KKASSERT(port->mpu_td == td);
 718     crit_enter_quick(td);
 719     while ((msg = _lwkt_pollmsg(port)) == NULL) {
 720         port->mp_flags |= MSGPORTF_WAITING;
 721         error = lwkt_sleep("waitport", flags);
 722         port->mp_flags &= ~MSGPORTF_WAITING;
 723         if (error)
 724             goto done;
 725     }
 726     _lwkt_pullmsg(port, msg);
 727 done:
 728     crit_exit_quick(td);
 729     return(msg);
 730 }
 731
 732 /************************************************************************
 733  *                         SPIN PORT BACKEND                            *
 734  ************************************************************************
 735  *
 736  * This backend uses spinlocks instead of making assumptions about which
 737  * thread is accessing the port.  It must be used when a port is not owned
 738  * by a particular thread.  This is less optimal then thread ports but
 739  * you don't have a choice if there are multiple threads accessing the port.
 740  *
 741  * Note on MSGPORTF_WAITING - because there may be multiple threads blocked
 742  * on the message port, it is the responsibility of the code doing the
 743  * wakeup to clear this flag rather then the blocked threads.  Some
 744  * superfluous wakeups may occur, which is ok.
 745  *
 746  * XXX synchronous message wakeups are not current optimized.
 747  */
 748
 749 static
 750 void *
 751 lwkt_spin_getport(lwkt_port_t port)
 752 {
 753     lwkt_msg_t msg;
 754
 755     spin_lock(&port->mpu_spin);
 756     if ((msg = _lwkt_pollmsg(port)) != NULL)
 757         _lwkt_pullmsg(port, msg);
 758     spin_unlock(&port->mpu_spin);
 759     return(msg);
 760 }
 761
 762 static
 763 int
 764 lwkt_spin_putport(lwkt_port_t port, lwkt_msg_t msg)
 765 {
 766     int dowakeup;
 767
 768     KKASSERT((msg->ms_flags & (MSGF_DONE | MSGF_REPLY)) == 0);
 769
 770     msg->ms_target_port = port;
 771     spin_lock(&port->mpu_spin);
 772     _lwkt_pushmsg(port, msg);
 773     dowakeup = 0;
 774     if (port->mp_flags & MSGPORTF_WAITING) {
 775         port->mp_flags &= ~MSGPORTF_WAITING;
 776         dowakeup = 1;
 777     }
 778     spin_unlock(&port->mpu_spin);
 779     if (dowakeup)
 780         wakeup(port);
 781     return (EASYNC);
 782 }
 783
 784 static
 785 int
 786 lwkt_spin_waitmsg(lwkt_msg_t msg, int flags)
 787 {
 788     lwkt_port_t port;
 789     int sentabort;
 790     int error;
 791
 792     KASSERT((msg->ms_flags & MSGF_DROPABLE) == 0,
 793             ("can't wait dropable message"));
 794
 795     if ((msg->ms_flags & MSGF_DONE) == 0) {
 796         port = msg->ms_reply_port;
 797         sentabort = 0;
 798         spin_lock(&port->mpu_spin);
 799         while ((msg->ms_flags & MSGF_DONE) == 0) {
 800             void *won;
 801
 802             /*
 803              * If message was sent synchronously from the beginning
 804              * the wakeup will be on the message structure, else it
 805              * will be on the port structure.
 806              */
 807             if (msg->ms_flags & MSGF_SYNC) {
 808                 won = msg;
 809                 atomic_set_int(&msg->ms_flags, MSGF_WAITING);
 810             } else {
 811                 won = port;
 812                 port->mp_flags |= MSGPORTF_WAITING;
 813             }
 814
 815             /*
 816              * Only messages which support abort can be interrupted.
 817              * We must still wait for message completion regardless.
 818              */
 819             if ((flags & PCATCH) && sentabort == 0) {
 820                 error = ssleep(won, &port->mpu_spin, PCATCH, "waitmsg", 0);
 821                 if (error) {
 822                     sentabort = error;
 823                     spin_unlock(&port->mpu_spin);
 824                     lwkt_abortmsg(msg);
 825                     spin_lock(&port->mpu_spin);
 826                 }
 827             } else {
 828                 error = ssleep(won, &port->mpu_spin, 0, "waitmsg", 0);
 829             }
 830             /* see note at the top on the MSGPORTF_WAITING flag */
 831         }
 832         /*
 833          * Turn EINTR into ERESTART if the signal indicates.
 834          */
 835         if (sentabort && msg->ms_error == EINTR)
 836             msg->ms_error = sentabort;
 837         if (msg->ms_flags & MSGF_QUEUED)
 838             _lwkt_pullmsg(port, msg);
 839         spin_unlock(&port->mpu_spin);
 840     } else {
 841         if (msg->ms_flags & MSGF_QUEUED) {
 842             port = msg->ms_reply_port;
 843             spin_lock(&port->mpu_spin);
 844             _lwkt_pullmsg(port, msg);
 845             spin_unlock(&port->mpu_spin);
 846         }
 847     }
 848     return(msg->ms_error);
 849 }
 850
 851 static
 852 void *
 853 lwkt_spin_waitport(lwkt_port_t port, int flags)
 854 {
 855     lwkt_msg_t msg;
 856     int error;
 857
 858     spin_lock(&port->mpu_spin);
 859     while ((msg = _lwkt_pollmsg(port)) == NULL) {
 860         port->mp_flags |= MSGPORTF_WAITING;
 861         error = ssleep(port, &port->mpu_spin, flags, "waitport", 0);
 862         /* see note at the top on the MSGPORTF_WAITING flag */
 863         if (error) {
 864             spin_unlock(&port->mpu_spin);
 865             return(NULL);
 866         }
 867     }
 868     _lwkt_pullmsg(port, msg);
 869     spin_unlock(&port->mpu_spin);
 870     return(msg);
 871 }
 872
 873 static
 874 void
 875 lwkt_spin_replyport(lwkt_port_t port, lwkt_msg_t msg)
 876 {
 877     int dowakeup;
 878
 879     KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == 0);
 880
 881     if (msg->ms_flags & MSGF_SYNC) {
 882         /*
 883          * If a synchronous completion has been requested, just wakeup
 884          * the message without bothering to queue it to the target port.
 885          */
 886         spin_lock(&port->mpu_spin);
 887         msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
 888         dowakeup = 0;
 889         if (msg->ms_flags & MSGF_WAITING) {
 890                 atomic_clear_int(&msg->ms_flags, MSGF_WAITING);
 891                 dowakeup = 1;
 892         }
 893         spin_unlock(&port->mpu_spin);
 894         if (dowakeup)
 895                 wakeup(msg);
 896     } else {
 897         /*
 898          * If an asynchronous completion has been requested the message
 899          * must be queued to the reply port.
 900          */
 901         spin_lock(&port->mpu_spin);
 902         _lwkt_enqueue_reply(port, msg);
 903         dowakeup = 0;
 904         if (port->mp_flags & MSGPORTF_WAITING) {
 905             port->mp_flags &= ~MSGPORTF_WAITING;
 906             dowakeup = 1;
 907         }
 908         spin_unlock(&port->mpu_spin);
 909         if (dowakeup)
 910             wakeup(port);
 911     }
 912 }
 913
 914 /*
 915  * lwkt_spin_dropmsg() - Backend to lwkt_dropmsg()
 916  *
 917  * This function could _only_ be used when caller is in the same thread
 918  * as the message's target port owner thread.
 919  */
 920 static void
 921 lwkt_spin_dropmsg(lwkt_port_t port, lwkt_msg_t msg)
 922 {
 923     KASSERT(port->mpu_td == curthread,
 924             ("message could only be dropped in the same thread "
 925              "as the message target port thread\n"));
 926     spin_lock(&port->mpu_spin);
 927     _lwkt_pullmsg(port, msg);
 928     msg->ms_flags |= MSGF_DONE;
 929     spin_unlock(&port->mpu_spin);
 930 }
 931
 932 /************************************************************************
 933  *                        SERIALIZER PORT BACKEND                       *
 934  ************************************************************************
 935  *
 936  * This backend uses serializer to protect port accessing.  Callers are
 937  * assumed to have serializer held.  This kind of port is usually created
 938  * by network device driver along with _one_ lwkt thread to pipeline
 939  * operations which may temporarily release serializer.
 940  *
 941  * Implementation is based on SPIN PORT BACKEND.
 942  */
 943
 944 static
 945 void *
 946 lwkt_serialize_getport(lwkt_port_t port)
 947 {
 948     lwkt_msg_t msg;
 949
 950     ASSERT_SERIALIZED(port->mpu_serialize);
 951
 952     if ((msg = _lwkt_pollmsg(port)) != NULL)
 953         _lwkt_pullmsg(port, msg);
 954     return(msg);
 955 }
 956
 957 static
 958 int
 959 lwkt_serialize_putport(lwkt_port_t port, lwkt_msg_t msg)
 960 {
 961     KKASSERT((msg->ms_flags & (MSGF_DONE | MSGF_REPLY)) == 0);
 962     ASSERT_SERIALIZED(port->mpu_serialize);
 963
 964     msg->ms_target_port = port;
 965     _lwkt_pushmsg(port, msg);
 966     if (port->mp_flags & MSGPORTF_WAITING) {
 967         port->mp_flags &= ~MSGPORTF_WAITING;
 968         wakeup(port);
 969     }
 970     return (EASYNC);
 971 }
 972
 973 static
 974 int
 975 lwkt_serialize_waitmsg(lwkt_msg_t msg, int flags)
 976 {
 977     lwkt_port_t port;
 978     int sentabort;
 979     int error;
 980
 981     KASSERT((msg->ms_flags & MSGF_DROPABLE) == 0,
 982             ("can't wait dropable message"));
 983
 984     if ((msg->ms_flags & MSGF_DONE) == 0) {
 985         port = msg->ms_reply_port;
 986
 987         ASSERT_SERIALIZED(port->mpu_serialize);
 988
 989         sentabort = 0;
 990         while ((msg->ms_flags & MSGF_DONE) == 0) {
 991             void *won;
 992
 993             /*
 994              * If message was sent synchronously from the beginning
 995              * the wakeup will be on the message structure, else it
 996              * will be on the port structure.
 997              */
 998             if (msg->ms_flags & MSGF_SYNC) {
 999                 won = msg;
1000             } else {
1001                 won = port;
1002                 port->mp_flags |= MSGPORTF_WAITING;
1003             }
1004
1005             /*
1006              * Only messages which support abort can be interrupted.
1007              * We must still wait for message completion regardless.
1008              */
1009             if ((flags & PCATCH) && sentabort == 0) {
1010                 error = zsleep(won, port->mpu_serialize, PCATCH, "waitmsg", 0);
1011                 if (error) {
1012                     sentabort = error;
1013                     lwkt_serialize_exit(port->mpu_serialize);
1014                     lwkt_abortmsg(msg);
1015                     lwkt_serialize_enter(port->mpu_serialize);
1016                 }
1017             } else {
1018                 error = zsleep(won, port->mpu_serialize, 0, "waitmsg", 0);
1019             }
1020             /* see note at the top on the MSGPORTF_WAITING flag */
1021         }
1022         /*
1023          * Turn EINTR into ERESTART if the signal indicates.
1024          */
1025         if (sentabort && msg->ms_error == EINTR)
1026             msg->ms_error = sentabort;
1027         if (msg->ms_flags & MSGF_QUEUED)
1028             _lwkt_pullmsg(port, msg);
1029     } else {
1030         if (msg->ms_flags & MSGF_QUEUED) {
1031             port = msg->ms_reply_port;
1032
1033             ASSERT_SERIALIZED(port->mpu_serialize);
1034             _lwkt_pullmsg(port, msg);
1035         }
1036     }
1037     return(msg->ms_error);
1038 }
1039
1040 static
1041 void *
1042 lwkt_serialize_waitport(lwkt_port_t port, int flags)
1043 {
1044     lwkt_msg_t msg;
1045     int error;
1046
1047     ASSERT_SERIALIZED(port->mpu_serialize);
1048
1049     while ((msg = _lwkt_pollmsg(port)) == NULL) {
1050         port->mp_flags |= MSGPORTF_WAITING;
1051         error = zsleep(port, port->mpu_serialize, flags, "waitport", 0);
1052         /* see note at the top on the MSGPORTF_WAITING flag */
1053         if (error)
1054             return(NULL);
1055     }
1056     _lwkt_pullmsg(port, msg);
1057     return(msg);
1058 }
1059
1060 static
1061 void
1062 lwkt_serialize_replyport(lwkt_port_t port, lwkt_msg_t msg)
1063 {
1064     KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == 0);
1065     ASSERT_SERIALIZED(port->mpu_serialize);
1066
1067     if (msg->ms_flags & MSGF_SYNC) {
1068         /*
1069          * If a synchronous completion has been requested, just wakeup
1070          * the message without bothering to queue it to the target port.
1071          */
1072         msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
1073         wakeup(msg);
1074     } else {
1075         /*
1076          * If an asynchronous completion has been requested the message
1077          * must be queued to the reply port.
1078          */
1079         _lwkt_enqueue_reply(port, msg);
1080         if (port->mp_flags & MSGPORTF_WAITING) {
1081             port->mp_flags &= ~MSGPORTF_WAITING;
1082             wakeup(port);
1083         }
1084     }
1085 }
1086
1087 /************************************************************************
1088  *                   PANIC AND SPECIAL PORT FUNCTIONS                   *
1089  ************************************************************************/
1090
1091 /*
1092  * You can point a port's reply vector at this function if you just want
1093  * the message marked done, without any queueing or signaling.  This is
1094  * often used for structure-embedded messages.
1095  */
1096 static
1097 void
1098 lwkt_null_replyport(lwkt_port_t port, lwkt_msg_t msg)
1099 {
1100     msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
1101 }
1102
1103 static
1104 void *
1105 lwkt_panic_getport(lwkt_port_t port)
1106 {
1107     panic("lwkt_getport() illegal on port %p", port);
1108 }
1109
1110 static
1111 int
1112 lwkt_panic_putport(lwkt_port_t port, lwkt_msg_t msg)
1113 {
1114     panic("lwkt_begin/do/sendmsg() illegal on port %p msg %p", port, msg);
1115 }
1116
1117 static
1118 int
1119 lwkt_panic_waitmsg(lwkt_msg_t msg, int flags)
1120 {
1121     panic("port %p msg %p cannot be waited on", msg->ms_reply_port, msg);
1122 }
1123
1124 static
1125 void *
1126 lwkt_panic_waitport(lwkt_port_t port, int flags)
1127 {
1128     panic("port %p cannot be waited on", port);
1129 }
1130
1131 static
1132 void
1133 lwkt_panic_replyport(lwkt_port_t port, lwkt_msg_t msg)
1134 {
1135     panic("lwkt_replymsg() is illegal on port %p msg %p", port, msg);
1136 }
1137
1138 static
1139 void
1140 lwkt_panic_dropmsg(lwkt_port_t port, lwkt_msg_t msg)
1141 {
1142     panic("lwkt_dropmsg() is illegal on port %p msg %p", port, msg);
1143 }