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.19 2004/04/15 00:50:03 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).
 103  *
 104  *      NOTE: you cannot safely request an abort until lwkt_sendmsg() returns
 105  *      to the caller.
 106  */
 107 void
 108 lwkt_sendmsg(lwkt_port_t port, lwkt_msg_t msg)
 109 {
 110     int error;
 111
 112     msg->ms_flags |= MSGF_ASYNC;
 113     msg->ms_flags &= ~(MSGF_REPLY1 | MSGF_REPLY2 | MSGF_QUEUED);
 114     KKASSERT(msg->ms_reply_port != NULL);
 115     msg->ms_abort_port = msg->ms_reply_port;
 116     if ((error = lwkt_beginmsg(port, msg)) != EASYNC) {
 117         lwkt_replymsg(msg, error);
 118     }
 119 }
 120
 121 /*
 122  * lwkt_domsg()
 123  *
 124  *      Send a message synchronously.  This function requests synchronous
 125  *      completion and calls lwkt_beginmsg().  If the target port decides to
 126  *      run the message asynchronously this function will block waiting for
 127  *      the message to complete.  Since MSGF_ASYNC is not set the target
 128  *      will not attempt to queue the reply to a reply port but will simply
 129  *      wake up anyone waiting on the message.
 130  *
 131  *      A synchronous error code is always returned.
 132  *
 133  *      The message's ms_cmd must be initialized, and its ms_flags must be
 134  *      at least zero'd out.  lwkt_domsg() will initialize the message's
 135  *      ms_abort_port (abort chasing port).
 136  *
 137  *      NOTE: you cannot safely request an abort until lwkt_domsg() blocks.
 138  *      XXX this probably needs some work.
 139  */
 140 int
 141 lwkt_domsg(lwkt_port_t port, lwkt_msg_t msg)
 142 {
 143     int error;
 144
 145     msg->ms_flags &= ~(MSGF_ASYNC | MSGF_REPLY1 | MSGF_REPLY2 | MSGF_QUEUED);
 146     KKASSERT(msg->ms_reply_port != NULL);
 147     msg->ms_abort_port = msg->ms_reply_port;
 148     if ((error = lwkt_beginmsg(port, msg)) == EASYNC) {
 149         error = lwkt_waitmsg(msg);
 150     }
 151     return(error);
 152 }
 153
 154 /************************************************************************
 155  *                              PORT FUNCTIONS                          *
 156  ************************************************************************/
 157
 158 /*
 159  * lwkt_initport()
 160  *
 161  *      Initialize a port for use and assign it to the specified thread.
 162  */
 163 void
 164 lwkt_initport(lwkt_port_t port, thread_t td)
 165 {
 166     bzero(port, sizeof(*port));
 167     TAILQ_INIT(&port->mp_msgq);
 168     port->mp_td = td;
 169     port->mp_putport = lwkt_default_putport;
 170     port->mp_waitport =  lwkt_default_waitport;
 171     port->mp_replyport = lwkt_default_replyport;
 172     port->mp_abortport = lwkt_default_abortport;
 173 }
 174
 175 /*
 176  * lwkt_getport()
 177  *
 178  *      Retrieve the next message from the port's message queue, return NULL
 179  *      if no messages are pending.  Note that callers CANNOT use the
 180  *      MSGF_ABORTED flag as a litmus test to determine if a message
 181  *      was aborted.  The flag only indicates that an abort was requested.
 182  *      The message's error code will indicate whether an abort occured
 183  *      (typically by returning EINTR).
 184  *
 185  *      Note that once a message has been dequeued it is subject to being
 186  *      requeued via an IPI based abort request if it is not marked MSGF_DONE.
 187  *
 188  *      The calling thread MUST own the port.
 189  */
 190 void *
 191 lwkt_getport(lwkt_port_t port)
 192 {
 193     lwkt_msg_t msg;
 194
 195     KKASSERT(port->mp_td == curthread);
 196
 197     crit_enter_quick(port->mp_td);
 198     if ((msg = TAILQ_FIRST(&port->mp_msgq)) != NULL) {
 199         TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 200         msg->ms_flags &= ~MSGF_QUEUED;
 201     }
 202     crit_exit_quick(port->mp_td);
 203     return(msg);
 204 }
 205
 206 /*
 207  * This inline helper function completes processing of a reply from an
 208  * unknown cpu context.
 209  *
 210  * The message is being returned to the specified port.  The port is
 211  * owned by the mp_td thread.  If we are on the same cpu as the mp_td
 212  * thread we can trivially queue the message to the messageq and schedule
 213  * the target thread, otherwise we have to send an ipi message to the
 214  * correct cpu.
 215  *
 216  * This inline must be entered with a critical section already held.
 217  * Note that the IPIQ callback function (*_remote) is entered with a
 218  * critical section already held, and we obtain one in lwkt_replyport().
 219  */
 220 static __inline
 221 void
 222 _lwkt_replyport(lwkt_port_t port, lwkt_msg_t msg, int force)
 223 {
 224     thread_t td = port->mp_td;
 225
 226     if (force || td->td_gd == mycpu) {
 227         /*
 228          * If an abort is racing us we cannot queue the reply now, the
 229          * abort code will have to do it when it catches up.
 230          */
 231         if (msg->ms_abort_port == port) {
 232             KKASSERT((msg->ms_flags & MSGF_QUEUED) == 0);
 233             TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 234             msg->ms_flags |= MSGF_DONE | MSGF_QUEUED | MSGF_REPLY2;
 235             if (port->mp_flags & MSGPORTF_WAITING)
 236                 lwkt_schedule(td);
 237         }
 238     } else {
 239         lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_replyport_remote, msg);
 240     }
 241 }
 242
 243 /*
 244  * This function completes reply processing for the default case in the
 245  * context of the originating cpu.
 246  */
 247 static
 248 void
 249 lwkt_replyport_remote(lwkt_msg_t msg)
 250 {
 251     _lwkt_replyport(msg->ms_reply_port, msg, 1);
 252 }
 253
 254 /*
 255  * This function is called in the context of the target to reply a message.
 256  * Note that the lwkt_replymsg() inline has already set MSGF_REPLY1 and
 257  * entered a critical section for us.
 258  */
 259 void
 260 lwkt_default_replyport(lwkt_port_t port, lwkt_msg_t msg)
 261 {
 262     if (msg->ms_flags & MSGF_ASYNC) {
 263         /*
 264          * An abort may have caught up to us while we were processing the
 265          * message.  If this occured we have to dequeue the message before
 266          * we can reply it.  If an abort occurs after we reply the MSGF_REPLY1
 267          * flag will prevent it from being queued to the target port.
 268          */
 269         if (msg->ms_flags & MSGF_QUEUED) {
 270             KKASSERT(msg->ms_flags & MSGF_ABORTED);
 271             TAILQ_REMOVE(&msg->ms_target_port->mp_msgq, msg, ms_node);
 272             msg->ms_flags &= ~MSGF_QUEUED;
 273         }
 274         _lwkt_replyport(port, msg, 0);
 275     } else {
 276         msg->ms_flags |= MSGF_DONE;
 277         if (port->mp_flags & MSGPORTF_WAITING)
 278             lwkt_schedule(port->mp_td);
 279     }
 280 }
 281
 282 /*
 283  * lwkt_default_putport()
 284  *
 285  *      This function is typically assigned to the mp_putport port vector.
 286  *
 287  *      Queue a message to the target port and wakeup the thread owning it.
 288  *      This function always returns EASYNC and may be assigned to a
 289  *      message port's mp_putport function vector.  Note that we must set
 290  *      MSGF_QUEUED prior to sending any IPIs in order to interlock against
 291  *      ABORT requests and other tests that might be performed.
 292  *
 293  *      The inline must be called from a critical section (the remote function
 294  *      is called from an IPI and will be in a critical section).
 295  */
 296 static
 297 __inline
 298 void
 299 _lwkt_putport(lwkt_port_t port, lwkt_msg_t msg, int force)
 300 {
 301     thread_t td = port->mp_td;
 302
 303     if (force || td->td_gd == mycpu) {
 304         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 305         if (port->mp_flags & MSGPORTF_WAITING)
 306             lwkt_schedule(td);
 307     } else {
 308         lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_putport_remote, msg);
 309     }
 310 }
 311
 312 static
 313 void
 314 lwkt_putport_remote(lwkt_msg_t msg)
 315 {
 316     _lwkt_putport(msg->ms_target_port, msg, 1);
 317 }
 318
 319 int
 320 lwkt_default_putport(lwkt_port_t port, lwkt_msg_t msg)
 321 {
 322     crit_enter();
 323     msg->ms_flags |= MSGF_QUEUED;       /* abort interlock */
 324     msg->ms_flags &= ~MSGF_DONE;
 325     msg->ms_target_port = port;
 326     _lwkt_putport(port, msg, 0);
 327     crit_exit();
 328     return(EASYNC);
 329 }
 330
 331 /*
 332  * lwkt_forwardmsg()
 333  *
 334  * Forward a message received on one port to another port.  The forwarding
 335  * function must deal with a pending abort but othewise essentially just
 336  * issues a putport to the target port.
 337  *
 338  * An abort may have two side effects:  First, the message may have been
 339  * requeued to the current target port.  If so, we must dequeue it before
 340  * we can forward it.
 341  */
 342 int
 343 lwkt_forwardmsg(lwkt_port_t port, lwkt_msg_t msg)
 344 {
 345     int error;
 346
 347     crit_enter();
 348     if (msg->ms_flags & MSGF_QUEUED) {
 349         KKASSERT(msg->ms_flags & MSGF_ABORTED);
 350         TAILQ_REMOVE(&msg->ms_target_port->mp_msgq, msg, ms_node);
 351         msg->ms_flags &= ~MSGF_QUEUED;
 352     }
 353     if ((error = port->mp_putport(port, msg)) != EASYNC)
 354         lwkt_replymsg(msg, error);
 355     crit_exit();
 356     return(error);
 357 }
 358
 359 /*
 360  * lwkt_abortmsg()
 361  *
 362  *      Aborting a message is a fairly complex task.  The first order of
 363  *      business is to get the message to the cpu that owns the target
 364  *      port, during which we may have to do some port chasing due to
 365  *      message forwarding operations.
 366  *
 367  *      NOTE!  Since an aborted message is requeued all message processing
 368  *      loops should check the MSGF_ABORTED flag.
 369  */
 370 static void lwkt_abortmsg_remote(lwkt_msg_t msg);
 371
 372 void
 373 lwkt_abortmsg(lwkt_msg_t msg)
 374 {
 375     lwkt_port_t port;
 376     thread_t td;
 377
 378     /*
 379      * A critical section protects us from reply IPIs on this cpu.  If
 380      * the message is marked done it has already been completely processed
 381      * and replied before we caled lwkt_abortmsg().
 382      */
 383     crit_enter();
 384     if ((msg->ms_flags & MSGF_DONE) == 0) {
 385         /*
 386          * Chase the message.  If REPLY1 is set the message has been replied
 387          * all the way back to the originator, otherwise it is sitting on
 388          * ms_target_port (but we can only complete processing if we are
 389          * on the same cpu as the selected port in order to avoid
 390          * SMP cache synchronization issues).
 391          *
 392          * When chasing through multiple ports ms_flags may not be
 393          * synchronized to the current cpu, but it WILL be synchronized
 394          * with regards to testing the MSGF_REPLY1 bit once we reach the
 395          * target port that made the reply and since the cpu owning
 396          * some port X stores the new port in ms_target_port if the message
 397          * is forwarded, the current port will only ever equal the target
 398          * port when we are on the correct cpu.
 399          */
 400         if (msg->ms_flags & MSGF_REPLY1)
 401             port = msg->ms_reply_port;
 402         else
 403             port = msg->ms_target_port;
 404         cpu_mb1();
 405         td = port->mp_td;
 406         if (td->td_gd != mycpu) {
 407             lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_abortmsg_remote, msg);
 408         } else {
 409             port->mp_abortport(port, msg);
 410         }
 411     }
 412     crit_exit();
 413 }
 414
 415 static
 416 void
 417 lwkt_abortmsg_remote(lwkt_msg_t msg)
 418 {
 419     lwkt_port_t port;
 420     thread_t td;
 421
 422     if (msg->ms_flags & MSGF_REPLY1)
 423         port = msg->ms_reply_port;
 424     else
 425         port = msg->ms_target_port;
 426     cpu_mb1();
 427     td = port->mp_td;
 428     if (td->td_gd != mycpu) {
 429         lwkt_send_ipiq(td->td_gd, (ipifunc_t)lwkt_abortmsg_remote, msg);
 430     } else {
 431         port->mp_abortport(port, msg);
 432     }
 433 }
 434
 435 /*
 436  * The mp_abortport function is called when the abort has finally caught up
 437  * to the target port or (if the message has been replied) the reply port.
 438  */
 439 void
 440 lwkt_default_abortport(lwkt_port_t port, lwkt_msg_t msg)
 441 {
 442     /*
 443      * Set ms_abort_port to ms_reply_port to indicate the completion of
 444      * the messaging chasing portion of the abort request.  Note that
 445      * the passed port is the port that we finally caught up to, not
 446      * necessarily the reply port.
 447      */
 448     msg->ms_abort_port = msg->ms_reply_port;
 449
 450     if (msg->ms_flags & MSGF_REPLY2) {
 451         /*
 452          * If REPLY2 is set we must have chased it all the way back to
 453          * the reply port.  We become responsible for
 454          */
 455         KKASSERT((msg->ms_flags & MSGF_QUEUED) == 0);
 456         KKASSERT(port == msg->ms_reply_port);
 457         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 458         msg->ms_flags |= MSGF_DONE | MSGF_QUEUED;
 459         if (port->mp_flags & MSGPORTF_WAITING)
 460             lwkt_schedule(port->mp_td);
 461     } else if ((msg->ms_flags & (MSGF_QUEUED|MSGF_REPLY1)) == 0) {
 462         msg->ms_flags |= MSGF_ABORTED | MSGF_QUEUED;
 463         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 464         if (port->mp_flags & MSGPORTF_WAITING)
 465             lwkt_schedule(port->mp_td);
 466     }
 467 }
 468
 469 /*
 470  * lwkt_default_waitport()
 471  *
 472  *      If msg is NULL, dequeue the next message from the port's message
 473  *      queue, block until a message is ready.  This function never
 474  *      returns NULL.
 475  *
 476  *      If msg is non-NULL, block until the requested message has been returned
 477  *      to the port then dequeue and return it.
 478  *
 479  *      Note that the API does not currently support multiple threads waiting
 480  *      on a port.  By virtue of owning the port it is controlled by our
 481  *      cpu and we can safely manipulate it's contents.
 482  */
 483 void *
 484 lwkt_default_waitport(lwkt_port_t port, lwkt_msg_t msg)
 485 {
 486     thread_t td = curthread;
 487     int sentabort;
 488
 489     KKASSERT(port->mp_td == td);
 490     crit_enter_quick(td);
 491     if (msg == NULL) {
 492         if ((msg = TAILQ_FIRST(&port->mp_msgq)) == NULL) {
 493             port->mp_flags |= MSGPORTF_WAITING;
 494             do {
 495                 lwkt_deschedule_self(td);
 496                 lwkt_switch();
 497             } while ((msg = TAILQ_FIRST(&port->mp_msgq)) == NULL);
 498             port->mp_flags &= ~MSGPORTF_WAITING;
 499         }
 500         TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 501         msg->ms_flags &= ~MSGF_QUEUED;
 502     } else {
 503         /*
 504          * If the message is marked done but not queued it has already been
 505          * pulled off the port and returned and we do not have to do anything.
 506          * Otherwise we do not own the message have to wait for message
 507          * completion.  Beware of cpu races if MSGF_DONE is not found to be
 508          * set!
 509          */
 510         if ((msg->ms_flags & (MSGF_DONE|MSGF_REPLY1)) != MSGF_DONE) {
 511             /*
 512              * We must own the reply port to safely mess with it's contents.
 513              */
 514             port = msg->ms_reply_port;
 515             KKASSERT(port->mp_td == curthread);
 516
 517             if ((msg->ms_flags & MSGF_DONE) == 0) {
 518                 port->mp_flags |= MSGPORTF_WAITING; /* saved by the BGL */
 519                 sentabort = 0;
 520                 do {
 521                     /*
 522                      * MSGF_PCATCH is only set by processes which wish to
 523                      * abort the message they are blocked on when a signal
 524                      * occurs.  Note that we still must wait for message
 525                      * completion after sending an abort request.
 526                      */
 527                     if (msg->ms_flags & MSGF_PCATCH) {
 528                         if (sentabort == 0 && CURSIG(port->mp_td->td_proc)) {
 529                             sentabort = 1;
 530                             lwkt_abortmsg(msg);
 531                         }
 532                     }
 533                     lwkt_deschedule_self(td);
 534                     lwkt_switch();
 535                 } while ((msg->ms_flags & MSGF_DONE) == 0);
 536                 port->mp_flags &= ~MSGPORTF_WAITING; /* saved by the BGL */
 537             }
 538             /*
 539              * We own the message now.
 540              */
 541             if (msg->ms_flags & MSGF_QUEUED) {
 542                 msg->ms_flags &= ~MSGF_QUEUED;
 543                 TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 544             }
 545         }
 546     }
 547     crit_exit_quick(td);
 548     return(msg);
 549 }
 550