sys/kern/lwkt_msgport.c

   1 /*
   2  * Copyright (c) 2003 Matthew Dillon <dillon@backplane.com>
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice, this list of conditions and the following disclaimer.
  10  * 2. Redistributions in binary form must reproduce the above copyright
  11  *    notice, this list of conditions and the following disclaimer in the
  12  *    documentation and/or other materials provided with the distribution.
  13  *
  14  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
  15  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  16  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  17  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
  18  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  19  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  20  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  21  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  22  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  23  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  24  * SUCH DAMAGE.
  25  *
  26  * NOTE! This file may be compiled for userland libraries as well as for
  27  * the kernel.
  28  *
  29  * $DragonFly: src/sys/kern/lwkt_msgport.c,v 1.20 2004/04/20 01:52:22 dillon Exp $
  30  */
  31
  32 #ifdef _KERNEL
  33
  34 #include <sys/param.h>
  35 #include <sys/systm.h>
  36 #include <sys/kernel.h>
  37 #include <sys/proc.h>
  38 #include <sys/rtprio.h>
  39 #include <sys/queue.h>
  40 #include <sys/sysctl.h>
  41 #include <sys/kthread.h>
  42 #include <sys/signalvar.h>
  43 #include <machine/cpu.h>
  44 #include <sys/lock.h>
  45
  46 #include <vm/vm.h>
  47 #include <vm/vm_param.h>
  48 #include <vm/vm_kern.h>
  49 #include <vm/vm_object.h>
  50 #include <vm/vm_page.h>
  51 #include <vm/vm_map.h>
  52 #include <vm/vm_pager.h>
  53 #include <vm/vm_extern.h>
  54 #include <vm/vm_zone.h>
  55
  56 #include <sys/thread2.h>
  57 #include <sys/msgport2.h>
  58
  59 #include <machine/stdarg.h>
  60 #include <machine/ipl.h>
  61 #include <machine/cpufunc.h>
  62 #ifdef SMP
  63 #include <machine/smp.h>
  64 #endif
  65
  66 #include <sys/malloc.h>
  67 MALLOC_DEFINE(M_LWKTMSG, "lwkt message", "lwkt message");
  68
  69 #else
  70
  71 #include <sys/stdint.h>
  72 #include <libcaps/thread.h>
  73 #include <sys/thread.h>
  74 #include <sys/msgport.h>
  75 #include <sys/errno.h>
  76 #include <libcaps/globaldata.h>
  77 #include <sys/thread2.h>
  78 #include <sys/msgport2.h>
  79 #include <string.h>
  80
  81 #endif /* _KERNEL */
  82
  83
  84 /************************************************************************
  85  *                              MESSAGE FUNCTIONS                       *
  86  ************************************************************************/
  87
  88 static void lwkt_replyport_remote(lwkt_msg_t msg);
  89 static void lwkt_putport_remote(lwkt_msg_t msg);
  90
  91 /*
  92  * lwkt_sendmsg()
  93  *
  94  *      Send a message asynchronously.  This function requests asynchronous
  95  *      completion and calls lwkt_beginmsg().  If the target port decides to
  96  *      run the message synchronously this function will automatically queue
  97  *      the message to the current thread's message queue to present a
  98  *      consistent interface to the caller.
  99  *
 100  *      The message's ms_cmd must be initialized and its ms_flags must
 101  *      be zero'd out.  lwkt_sendmsg() will initialize the ms_abort_port
 102  *      (abort chasing port).  If abort is supported, ms_abort must also be
 103  *      initialized.
 104  *
 105  *      NOTE: you cannot safely request an abort until lwkt_sendmsg() returns
 106  *      to the caller.
 107  */
 108 void
 109 lwkt_sendmsg(lwkt_port_t port, lwkt_msg_t msg)
 110 {
 111     int error;
 112
 113     msg->ms_flags |= MSGF_ASYNC;
 114     msg->ms_flags &= ~(MSGF_REPLY1 | MSGF_REPLY2 | MSGF_QUEUED | \
 115                         MSGF_ABORTED | MSGF_RETRIEVED);
 116     KKASSERT(msg->ms_reply_port != NULL);
 117     msg->ms_abort_port = msg->ms_reply_port;
 118     if ((error = lwkt_beginmsg(port, msg)) != EASYNC) {
 119         lwkt_replymsg(msg, error);
 120     }
 121 }
 122
 123 /*
 124  * lwkt_domsg()
 125  *
 126  *      Send a message synchronously.  This function requests synchronous
 127  *      completion and calls lwkt_beginmsg().  If the target port decides to
 128  *      run the message asynchronously this function will block waiting for
 129  *      the message to complete.  Since MSGF_ASYNC is not set the target
 130  *      will not attempt to queue the reply to a reply port but will simply
 131  *      wake up anyone waiting on the message.
 132  *
 133  *      A synchronous error code is always returned.
 134  *
 135  *      The message's ms_cmd must be initialized, and its ms_flags must be
 136  *      at least zero'd out.  lwkt_domsg() will initialize the message's
 137  *      ms_abort_port (abort chasing port).  If abort is supported, ms_abort
 138  *      must also be initialized.
 139  *
 140  *      NOTE: you cannot safely request an abort until lwkt_domsg() blocks.
 141  *      XXX this probably needs some work.
 142  */
 143 int
 144 lwkt_domsg(lwkt_port_t port, lwkt_msg_t msg)
 145 {
 146     int error;
 147
 148     msg->ms_flags &= ~(MSGF_ASYNC | MSGF_REPLY1 | MSGF_REPLY2 | \
 149                         MSGF_QUEUED | MSGF_ABORTED | MSGF_RETRIEVED);
 150     KKASSERT(msg->ms_reply_port != NULL);
 151     msg->ms_abort_port = msg->ms_reply_port;
 152     if ((error = lwkt_beginmsg(port, msg)) == EASYNC) {
 153         error = lwkt_waitmsg(msg);
 154     }
 155     return(error);
 156 }
 157
 158 /************************************************************************
 159  *                              PORT FUNCTIONS                          *
 160  ************************************************************************/
 161
 162 /*
 163  * lwkt_initport()
 164  *
 165  *      Initialize a port for use and assign it to the specified thread.
 166  */
 167 void
 168 lwkt_initport(lwkt_port_t port, thread_t td)
 169 {
 170     bzero(port, sizeof(*port));
 171     TAILQ_INIT(&port->mp_msgq);
 172     port->mp_td = td;
 173     port->mp_putport = lwkt_default_putport;
 174     port->mp_waitport =  lwkt_default_waitport;
 175     port->mp_replyport = lwkt_default_replyport;
 176     port->mp_abortport = lwkt_default_abortport;
 177 }
 178
 179 /*
 180  * lwkt_getport()
 181  *
 182  *      Retrieve the next message from the port's message queue, return NULL
 183  *      if no messages are pending.  Note that callers CANNOT use the
 184  *      MSGF_ABORTED flag as a litmus test to determine if a message
 185  *      was aborted.  The flag only indicates that an abort was requested.
 186  *      The message's error code will indicate whether an abort occured
 187  *      (typically by returning EINTR).
 188  *
 189  *      Note that once a message has been dequeued it is subject to being
 190  *      requeued via an IPI based abort request if it is not marked MSGF_DONE.
 191  *
 192  *      If the message has been aborted we have to guarentee that abort
 193  *      semantics are properly followed.   The target port will always see
 194  *      the original message at least once, and if it does not reply the
 195  *      message before looping on its message port again it will then see
 196  *      the message again with ms_cmd set to ms_abort.
 197  *
 198  *      The calling thread MUST own the port.
 199  */
 200
 201 static __inline
 202 void
 203 _lwkt_pullmsg(lwkt_port_t port, lwkt_msg_t msg)
 204 {
 205     if ((msg->ms_flags & MSGF_ABORTED) == 0) {
 206         /*
 207          * normal case, remove and return the message.
 208          */
 209         TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 210         msg->ms_flags = (msg->ms_flags & ~MSGF_QUEUED) | MSGF_RETRIEVED;
 211     } else {
 212         if (msg->ms_flags & MSGF_RETRIEVED) {
 213             /*
 214              * abort case, message already returned once, remvoe and
 215              * return the aborted message a second time after setting
 216              * ms_cmd to ms_abort.
 217              */
 218             TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 219             msg->ms_flags &= ~MSGF_QUEUED;
 220             msg->ms_cmd = msg->ms_abort;
 221         } else {
 222             /*
 223              * abort case, abort races initial message retrieval.  The
 224              * message is returned normally but not removed from the
 225              * queue.  On the next loop the 'aborted' message will be
 226              * dequeued and returned.  Note that if the caller replies
 227              * to the message it will be dequeued (the abort becomes a
 228              * NOP).
 229              */
 230             msg->ms_flags |= MSGF_RETRIEVED;
 231         }
 232     }
 233 }
 234
 235 void *
 236 lwkt_getport(lwkt_port_t port)
 237 {
 238     lwkt_msg_t msg;
 239
 240     KKASSERT(port->mp_td == curthread);
 241
 242     crit_enter_quick(port->mp_td);
 243     if ((msg = TAILQ_FIRST(&port->mp_msgq)) != NULL)
 244         _lwkt_pullmsg(port, msg);
 245     crit_exit_quick(port->mp_td);
 246     return(msg);
 247 }
 248
 249 /*
 250  * This inline helper function completes processing of a reply from an
 251  * unknown cpu context.
 252  *
 253  * The message is being returned to the specified port.  The port is
 254  * owned by the mp_td thread.  If we are on the same cpu as the mp_td
 255  * thread we can trivially queue the message to the reply port and schedule
 256  * the target thread, otherwise we have to send an ipi message to the
 257  * correct cpu.
 258  *
 259  * This inline must be entered with a critical section already held.
 260  * Note that the IPIQ callback function (*_remote) is entered with a
 261  * critical section already held, and we obtain one in lwkt_replyport().
 262  */
 263 static __inline
 264 void
 265 _lwkt_replyport(lwkt_port_t port, lwkt_msg_t msg, int force)
 266 {
 267     thread_t td = port->mp_td;
 268
 269     if (force || td->td_gd == mycpu) {
 270         /*
 271          * We can only reply the message if the abort has caught up with us,
 272          * or if no abort was issued (same case).
 273          */
 274         if (msg->ms_abort_port == port) {
 275             KKASSERT((msg->ms_flags & MSGF_QUEUED) == 0);
 276             TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 277             msg->ms_flags |= MSGF_DONE | MSGF_QUEUED | MSGF_REPLY2;
 278             if (port->mp_flags & MSGPORTF_WAITING)
 279                 lwkt_schedule(td);
 280         }
 281     } else {
 282         lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_replyport_remote, msg);
 283     }
 284 }
 285
 286 /*
 287  * This function completes reply processing for the default case in the
 288  * context of the originating cpu.
 289  */
 290 static
 291 void
 292 lwkt_replyport_remote(lwkt_msg_t msg)
 293 {
 294     _lwkt_replyport(msg->ms_reply_port, msg, 1);
 295 }
 296
 297 /*
 298  * This function is called in the context of the target to reply a message.
 299  * Note that the lwkt_replymsg() inline has already set MSGF_REPLY1 and
 300  * entered a critical section for us.
 301  */
 302
 303 void
 304 lwkt_default_replyport(lwkt_port_t port, lwkt_msg_t msg)
 305 {
 306     crit_enter();
 307     msg->ms_flags |= MSGF_REPLY1;
 308     if (msg->ms_flags & MSGF_ASYNC) {
 309         /*
 310          * An abort may have caught up to us while we were processing the
 311          * message.  If this occured we have to dequeue the message from the
 312          * target port in the context of our current cpu before we can
 313          * finish replying it.
 314          *
 315          * If an abort occurs after we reply the MSGF_REPLY1 flag will
 316          * prevent it from being requeued to the target port.
 317          */
 318         if (msg->ms_flags & MSGF_QUEUED) {
 319             KKASSERT(msg->ms_flags & MSGF_ABORTED);
 320             TAILQ_REMOVE(&msg->ms_target_port->mp_msgq, msg, ms_node);
 321             msg->ms_flags &= ~MSGF_QUEUED;
 322         }
 323         _lwkt_replyport(port, msg, 0);
 324     } else {
 325         /*
 326          * Synchronously executed messages cannot be aborted and are just
 327          * marked done.  YYY MSGF_DONE should already be set, change flag set
 328          * to KKASSERT.
 329          */
 330         msg->ms_flags |= MSGF_DONE;
 331         if (port->mp_flags & MSGPORTF_WAITING)
 332             lwkt_schedule(port->mp_td);
 333     }
 334     crit_exit();
 335 }
 336
 337 /*
 338  * lwkt_default_putport()
 339  *
 340  *      This function is typically assigned to the mp_putport port vector.
 341  *
 342  *      Queue a message to the target port and wakeup the thread owning it.
 343  *      This function always returns EASYNC and may be assigned to a
 344  *      message port's mp_putport function vector.  Note that we must set
 345  *      MSGF_QUEUED prior to sending any IPIs in order to interlock against
 346  *      ABORT requests and other tests that might be performed.
 347  *
 348  *      The inline must be called from a critical section (the remote function
 349  *      is called from an IPI and will be in a critical section).
 350  */
 351 static
 352 __inline
 353 void
 354 _lwkt_putport(lwkt_port_t port, lwkt_msg_t msg, int force)
 355 {
 356     thread_t td = port->mp_td;
 357
 358     if (force || td->td_gd == mycpu) {
 359         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 360         if (port->mp_flags & MSGPORTF_WAITING)
 361             lwkt_schedule(td);
 362     } else {
 363         lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_putport_remote, msg);
 364     }
 365 }
 366
 367 static
 368 void
 369 lwkt_putport_remote(lwkt_msg_t msg)
 370 {
 371     _lwkt_putport(msg->ms_target_port, msg, 1);
 372 }
 373
 374 int
 375 lwkt_default_putport(lwkt_port_t port, lwkt_msg_t msg)
 376 {
 377     crit_enter();
 378     msg->ms_flags |= MSGF_QUEUED;       /* abort interlock */
 379     msg->ms_flags &= ~MSGF_DONE;
 380     msg->ms_target_port = port;
 381     _lwkt_putport(port, msg, 0);
 382     crit_exit();
 383     return(EASYNC);
 384 }
 385
 386 /*
 387  * lwkt_forwardmsg()
 388  *
 389  * Forward a message received on one port to another port.  The forwarding
 390  * function must deal with a pending abort but othewise essentially just
 391  * issues a putport to the target port.
 392  *
 393  * An abort may have two side effects:  First, the message may have been
 394  * requeued to the current target port.  If so, we must dequeue it before
 395  * we can forward it.
 396  */
 397 int
 398 lwkt_forwardmsg(lwkt_port_t port, lwkt_msg_t msg)
 399 {
 400     int error;
 401
 402     crit_enter();
 403     if (msg->ms_flags & MSGF_QUEUED) {
 404         KKASSERT(msg->ms_flags & MSGF_ABORTED);
 405         TAILQ_REMOVE(&msg->ms_target_port->mp_msgq, msg, ms_node);
 406         msg->ms_flags &= ~MSGF_QUEUED;
 407     }
 408     msg->ms_flags &= ~MSGF_RETRIEVED;
 409     if ((error = port->mp_putport(port, msg)) != EASYNC)
 410         lwkt_replymsg(msg, error);
 411     crit_exit();
 412     return(error);
 413 }
 414
 415 /*
 416  * lwkt_abortmsg()
 417  *
 418  *      Aborting a message is a fairly complex task.  The first order of
 419  *      business is to get the message to the cpu that owns the target
 420  *      port, during which we may have to do some port chasing due to
 421  *      message forwarding operations.
 422  *
 423  *      NOTE!  Since an aborted message is requeued all message processing
 424  *      loops should check the MSGF_ABORTED flag.
 425  */
 426 static void lwkt_abortmsg_remote(lwkt_msg_t msg);
 427
 428 void
 429 lwkt_abortmsg(lwkt_msg_t msg)
 430 {
 431     lwkt_port_t port;
 432     thread_t td;
 433
 434     /*
 435      * A critical section protects us from reply IPIs on this cpu.   We
 436      * can only abort messages that have not yet completed (DONE), are not
 437      * in the midst of being replied (REPLY1), and which support the
 438      * abort function (ABORTABLE).
 439      */
 440     crit_enter();
 441     if ((msg->ms_flags & (MSGF_DONE|MSGF_REPLY1|MSGF_ABORTABLE)) == MSGF_ABORTABLE) {
 442         /*
 443          * Chase the message.  If REPLY1 is set the message has been replied
 444          * all the way back to the originator, otherwise it is sitting on
 445          * ms_target_port (but we can only complete processing if we are
 446          * on the same cpu as the selected port in order to avoid
 447          * SMP cache synchronization issues).
 448          *
 449          * When chasing through multiple ports ms_flags may not be
 450          * synchronized to the current cpu, but it WILL be synchronized
 451          * with regards to testing the MSGF_REPLY1 bit once we reach the
 452          * target port that made the reply and since the cpu owning
 453          * some port X stores the new port in ms_target_port if the message
 454          * is forwarded, the current port will only ever equal the target
 455          * port when we are on the correct cpu.
 456          */
 457         if (msg->ms_flags & MSGF_REPLY1)
 458             port = msg->ms_reply_port;
 459         else
 460             port = msg->ms_target_port;
 461         cpu_mb1();
 462         td = port->mp_td;
 463         if (td->td_gd != mycpu) {
 464             lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_abortmsg_remote, msg);
 465         } else {
 466             port->mp_abortport(port, msg);
 467         }
 468     }
 469     crit_exit();
 470 }
 471
 472 static
 473 void
 474 lwkt_abortmsg_remote(lwkt_msg_t msg)
 475 {
 476     lwkt_port_t port;
 477     thread_t td;
 478
 479     if (msg->ms_flags & MSGF_REPLY1)
 480         port = msg->ms_reply_port;
 481     else
 482         port = msg->ms_target_port;
 483     cpu_mb1();
 484     td = port->mp_td;
 485     if (td->td_gd != mycpu) {
 486         lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_abortmsg_remote, msg);
 487     } else {
 488         port->mp_abortport(port, msg);
 489     }
 490 }
 491
 492 /*
 493  * The mp_abortport function is called when the abort has finally caught up
 494  * to the target port or (if the message has been replied) the reply port.
 495  */
 496 void
 497 lwkt_default_abortport(lwkt_port_t port, lwkt_msg_t msg)
 498 {
 499     /*
 500      * Set ms_abort_port to ms_reply_port to indicate the completion of
 501      * the messaging chasing portion of the abort request.  Note that
 502      * the passed port is the port that we finally caught up to, not
 503      * necessarily the reply port.
 504      */
 505     msg->ms_abort_port = msg->ms_reply_port;
 506
 507     if (msg->ms_flags & MSGF_REPLY2) {
 508         /*
 509          * If REPLY2 is set we must have chased it all the way back to
 510          * the reply port, but the replyport code has not queued the message
 511          * (because it was waiting for the abort to catch up).  We become
 512          * responsible for queueing the message to the reply port.
 513          */
 514         KKASSERT((msg->ms_flags & MSGF_QUEUED) == 0);
 515         KKASSERT(port == msg->ms_reply_port);
 516         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 517         msg->ms_flags |= MSGF_DONE | MSGF_QUEUED;
 518         if (port->mp_flags & MSGPORTF_WAITING)
 519             lwkt_schedule(port->mp_td);
 520     } else if ((msg->ms_flags & (MSGF_QUEUED|MSGF_REPLY1)) == 0) {
 521         /*
 522          * Abort on the target port.  The message has not yet been replied
 523          * and must be requeued to the target port.
 524          */
 525         msg->ms_flags |= MSGF_ABORTED | MSGF_QUEUED;
 526         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 527         if (port->mp_flags & MSGPORTF_WAITING)
 528             lwkt_schedule(port->mp_td);
 529     } else if ((msg->ms_flags & MSGF_REPLY1) == 0) {
 530         /*
 531          * The message has not yet been retrieved by the target port, set
 532          * MSGF_ABORTED so the target port can requeue the message abort after
 533          * retrieving it.
 534          */
 535         msg->ms_flags |= MSGF_ABORTED;
 536     }
 537 }
 538
 539 /*
 540  * lwkt_default_waitport()
 541  *
 542  *      If msg is NULL, dequeue the next message from the port's message
 543  *      queue, block until a message is ready.  This function never
 544  *      returns NULL.
 545  *
 546  *      If msg is non-NULL, block until the requested message has been returned
 547  *      to the port then dequeue and return it.  DO NOT USE THIS TO WAIT FOR
 548  *      INCOMING REQUESTS, ONLY USE THIS TO WAIT FOR REPLIES.
 549  *
 550  *      Note that the API does not currently support multiple threads waiting
 551  *      on a port.  By virtue of owning the port it is controlled by our
 552  *      cpu and we can safely manipulate it's contents.
 553  */
 554 void *
 555 lwkt_default_waitport(lwkt_port_t port, lwkt_msg_t msg)
 556 {
 557     thread_t td = curthread;
 558     int sentabort;
 559
 560     KKASSERT(port->mp_td == td);
 561     crit_enter_quick(td);
 562     if (msg == NULL) {
 563         if ((msg = TAILQ_FIRST(&port->mp_msgq)) == NULL) {
 564             port->mp_flags |= MSGPORTF_WAITING;
 565             do {
 566                 lwkt_deschedule_self(td);
 567                 lwkt_switch();
 568             } while ((msg = TAILQ_FIRST(&port->mp_msgq)) == NULL);
 569             port->mp_flags &= ~MSGPORTF_WAITING;
 570         }
 571         _lwkt_pullmsg(port, msg);
 572     } else {
 573         /*
 574          * If a message is not marked done, or if it is queued, we have work
 575          * to do.  Note that MSGF_DONE is always set in the context of the
 576          * reply port's cpu.
 577          */
 578         if ((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) != MSGF_DONE) {
 579             /*
 580              * We must own the reply port to safely mess with it's contents.
 581              */
 582             port = msg->ms_reply_port;
 583             KKASSERT(port->mp_td == td);
 584
 585             if ((msg->ms_flags & MSGF_DONE) == 0) {
 586                 port->mp_flags |= MSGPORTF_WAITING; /* saved by the BGL */
 587                 sentabort = 0;
 588                 do {
 589                     /*
 590                      * MSGF_PCATCH is only set by processes which wish to
 591                      * abort the message they are blocked on when a signal
 592                      * occurs.  Note that we still must wait for message
 593                      * completion after sending an abort request.
 594                      */
 595                     if (msg->ms_flags & MSGF_PCATCH) {
 596                         if (sentabort == 0 && CURSIG(port->mp_td->td_proc)) {
 597                             sentabort = 1;
 598                             lwkt_abortmsg(msg);
 599                         }
 600                     }
 601                     lwkt_deschedule_self(td);
 602                     lwkt_switch();
 603                 } while ((msg->ms_flags & MSGF_DONE) == 0);
 604                 port->mp_flags &= ~MSGPORTF_WAITING; /* saved by the BGL */
 605             }
 606             /*
 607              * We own the message now.
 608              */
 609             if (msg->ms_flags & MSGF_QUEUED) {
 610                 msg->ms_flags &= ~MSGF_QUEUED;
 611                 TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 612             }
 613         }
 614     }
 615     crit_exit_quick(td);
 616     return(msg);
 617 }
 618