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  * $DragonFly: src/sys/kern/lwkt_msgport.c,v 1.2 2003/07/22 17:03:33 dillon Exp $
  27  */
  28
  29 #include <sys/param.h>
  30 #include <sys/systm.h>
  31 #include <sys/kernel.h>
  32 #include <sys/proc.h>
  33 #include <sys/rtprio.h>
  34 #include <sys/queue.h>
  35 #include <sys/sysctl.h>
  36 #include <sys/kthread.h>
  37 #include <machine/cpu.h>
  38 #include <sys/lock.h>
  39
  40 #include <vm/vm.h>
  41 #include <vm/vm_param.h>
  42 #include <vm/vm_kern.h>
  43 #include <vm/vm_object.h>
  44 #include <vm/vm_page.h>
  45 #include <vm/vm_map.h>
  46 #include <vm/vm_pager.h>
  47 #include <vm/vm_extern.h>
  48 #include <vm/vm_zone.h>
  49
  50 #include <sys/thread2.h>
  51 #include <sys/msgport2.h>
  52
  53 #include <machine/stdarg.h>
  54 #include <machine/ipl.h>
  55 #ifdef SMP
  56 #include <machine/smp.h>
  57 #endif
  58
  59
  60 /************************************************************************
  61  *                              MESSAGE FUNCTIONS                       *
  62  ************************************************************************/
  63
  64 static void lwkt_replyport_remote(lwkt_msg_t msg);
  65 static void lwkt_putport_remote(lwkt_msg_t msg);
  66 static void lwkt_abortport_remote(lwkt_port_t port);
  67
  68 /*
  69  * lwkt_sendmsg()
  70  *
  71  *      Send a message asynchronously.  This function requests asynchronous
  72  *      completion and calls lwkt_beginmsg().  If the target port decides to
  73  *      run the message synchronously this function will automatically queue
  74  *      the message to the current thread's message queue to present a
  75  *      consistent interface to the caller.
  76  *
  77  *      The message's ms_cmd must be initialized and its ms_flags must be
  78  *      at least zero'd out.  lwkt_sendmsg() will initialize the message's
  79  *      reply port to the current thread's built-in reply port.
  80  */
  81 void
  82 lwkt_sendmsg(lwkt_port_t port, lwkt_msg_t msg)
  83 {
  84     int error;
  85
  86     msg->ms_flags |= MSGF_ASYNC;
  87     msg->ms_flags &= ~(MSGF_REPLY | MSGF_QUEUED);
  88     msg->ms_reply_port = &curthread->td_msgport;
  89     msg->ms_abortreq = 0;
  90     if ((error = lwkt_beginmsg(port, msg)) != EINPROGRESS) {
  91         lwkt_replymsg(msg, error);
  92     }
  93 }
  94
  95 /*
  96  * lwkt_domsg()
  97  *
  98  *      Send a message synchronously.  This function requests synchronous
  99  *      completion and calls lwkt_beginmsg().  If the target port decides to
 100  *      run the message asynchronously this function will block waiting for
 101  *      the message to complete.  Since MSGF_ASYNC is not set the target
 102  *      will not attempt to queue the reply to a reply port but will simply
 103  *      wake up anyone waiting on the message.
 104  *
 105  *      A synchronous error code is always returned.
 106  *
 107  *      The message's ms_cmd must be initialized and its ms_flags must be
 108  *      at least zero'd out.  lwkt_domsg() will initialize the message's
 109  *      reply port to the current thread's built-in reply port.
 110  */
 111 int
 112 lwkt_domsg(lwkt_port_t port, lwkt_msg_t msg)
 113 {
 114     int error;
 115
 116     msg->ms_flags &= ~(MSGF_ASYNC | MSGF_REPLY | MSGF_QUEUED);
 117     msg->ms_reply_port = &curthread->td_msgport;
 118     msg->ms_abortreq = 0;
 119     if ((error = lwkt_beginmsg(port, msg)) == EINPROGRESS) {
 120         error = lwkt_waitmsg(msg);
 121     }
 122     return(error);
 123 }
 124
 125 /*
 126  * lwkt_waitmsg()
 127  *
 128  *      Wait for a message that we originated to complete, remove it from
 129  *      the reply queue if necessary, and return its error code.
 130  *
 131  *      This call may be used in virtually any situation, including for the
 132  *      case where you used lwkt_sendmsg() to initiate the message.
 133  *
 134  *      Note that we don't own the message any more so we cannot safely
 135  *      modify ms_flags, meaning we can't clear MSGF_ASYNC as an optimization.
 136  *      However, since any remote cpu replying to the message will IPI the
 137  *      message over to us for action, a critical section is sufficient to
 138  *      protect td_msgq.
 139  */
 140 int
 141 lwkt_waitmsg(lwkt_msg_t msg)
 142 {
 143     lwkt_port_t port;
 144
 145     /*
 146      * Done but not queued case (message was originally a synchronous request)
 147      */
 148     if ((msg->ms_flags & (MSGF_DONE|MSGF_REPLY)) == MSGF_DONE)
 149         return(msg->ms_error);
 150
 151     port = msg->ms_reply_port;
 152     KKASSERT(port->mp_td == curthread); /* for now */
 153     crit_enter();
 154     if ((msg->ms_flags & MSGF_DONE) == 0) {
 155         port->mp_flags |= MSGPORTF_WAITING;
 156         do {
 157             lwkt_deschedule_self();
 158             lwkt_switch();
 159         } while ((msg->ms_flags & MSGF_DONE) == 0);
 160         port->mp_flags &= ~MSGPORTF_WAITING;
 161     }
 162     /*
 163      * We own the message now.
 164      */
 165     if (msg->ms_flags & MSGF_QUEUED) {
 166         msg->ms_flags &= ~MSGF_QUEUED;
 167         TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 168     }
 169     crit_exit();
 170     return(msg->ms_error);
 171 }
 172
 173
 174 /************************************************************************
 175  *                              PORT FUNCTIONS                          *
 176  ************************************************************************/
 177
 178 /*
 179  * lwkt_initport()
 180  *
 181  *      Initialize a port for use and assign it to the specified thread.
 182  */
 183 void
 184 lwkt_init_port(lwkt_port_t port, thread_t td)
 185 {
 186     bzero(port, sizeof(*port));
 187     TAILQ_INIT(&port->mp_msgq);
 188     port->mp_td = td;
 189     port->mp_beginmsg = lwkt_putport;
 190     port->mp_abortmsg =  lwkt_abortport;
 191     port->mp_returnmsg = lwkt_replyport;
 192 }
 193
 194 /*
 195  * lwkt_replyport()
 196  *
 197  *      This function is typically assigned to the mp_replymsg port vector.
 198  *
 199  *      The message is being returned to the specified port.  The port is
 200  *      owned by the mp_td thread.  If we are on the same cpu as the mp_td
 201  *      thread we can trivially queue the message to the messageq, otherwise
 202  *      we have to send an ipi message to the correct cpu.   We then schedule
 203  *      the target thread.
 204  *
 205  *      If MSGF_ASYNC is not set we do not bother queueing the message, we
 206  *      just set the DONE bit.
 207  *
 208  *      Note that the IPIQ callback function (*_remote) is entered with a
 209  *      critical section already held.
 210  */
 211
 212 static __inline
 213 void
 214 _lwkt_replyport(lwkt_port_t port, lwkt_msg_t msg)
 215 {
 216     thread_t td = port->mp_td;
 217
 218     if (td->td_cpu == mycpu->gd_cpuid) {
 219         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 220         msg->ms_flags |= MSGF_DONE | MSGF_REPLY | MSGF_QUEUED;
 221         if (port->mp_flags & MSGPORTF_WAITING)
 222             lwkt_schedule(td);
 223     } else {
 224         lwkt_send_ipiq(td->td_cpu, (ipifunc_t)lwkt_replyport_remote, msg);
 225     }
 226 }
 227
 228 static
 229 void
 230 lwkt_replyport_remote(lwkt_msg_t msg)
 231 {
 232     _lwkt_replyport(msg->ms_reply_port, msg);
 233 }
 234
 235
 236 void
 237 lwkt_replyport(lwkt_port_t port, lwkt_msg_t msg)
 238 {
 239     crit_enter();
 240     if (msg->ms_flags & MSGF_ASYNC) {
 241         _lwkt_replyport(port, msg);
 242     } else {
 243         msg->ms_flags |= MSGF_DONE;
 244         if (port->mp_flags & MSGPORTF_WAITING)
 245             lwkt_schedule(port->mp_td);
 246     }
 247     crit_exit();
 248 }
 249
 250 /*
 251  * lwkt_putport()
 252  *
 253  *      This function is typically assigned to the mp_beginmsg port vector.
 254  *
 255  *      Queue a message to the target port and wakeup the thread owning it.
 256  *      This function always returns EINPROGRESS and may be assigned to a
 257  *      message port's mp_beginmsg function vector.
 258  */
 259 static
 260 __inline
 261 void
 262 _lwkt_putport(lwkt_port_t port, lwkt_msg_t msg)
 263 {
 264     thread_t td = port->mp_td;
 265
 266     if (td->td_cpu == mycpu->gd_cpuid) {
 267         TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
 268         msg->ms_flags |= MSGF_QUEUED;
 269         if (port->mp_flags & MSGPORTF_WAITING)
 270             lwkt_schedule(td);
 271     } else {
 272         msg->ms_target_port = port;
 273         lwkt_send_ipiq(td->td_cpu, (ipifunc_t)lwkt_putport_remote, msg);
 274     }
 275 }
 276
 277 static
 278 void
 279 lwkt_putport_remote(lwkt_msg_t msg)
 280 {
 281     _lwkt_putport(msg->ms_target_port, msg);
 282 }
 283
 284 int
 285 lwkt_putport(lwkt_port_t port, lwkt_msg_t msg)
 286 {
 287     crit_enter();
 288     _lwkt_putport(port, msg);
 289     crit_exit();
 290     return(EINPROGRESS);
 291 }
 292
 293 /*
 294  * lwkt_abortport()
 295  *
 296  *      This function is typically assigned to the mp_abortmsg port vector.
 297  *
 298  *      This function attempts to abort a message.  Aborts are always
 299  *      optional, so we could just do nothing if we wanted.  We get onto the
 300  *      cpu owning the port, check to see if the message is queued on the
 301  *      port's message queue, and remove and abort it if it is.  Otherwise
 302  *      we do nothing.
 303  *
 304  *      Note that we cannot safely use ms_target_port because the port might
 305  *      have forwarded the message on to some other port and we could race
 306  *      against that use of ms_target_port.
 307  */
 308 static
 309 __inline
 310 void
 311 _lwkt_abortport(lwkt_port_t port)
 312 {
 313     thread_t td = port->mp_td;
 314     lwkt_msg_t msg;
 315
 316     if (td->td_cpu == mycpu->gd_cpuid) {
 317 again:
 318         msg = TAILQ_FIRST(&port->mp_msgq);
 319         while (msg) {
 320             if (msg->ms_abortreq) {
 321                 TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 322                 msg->ms_flags &= ~MSGF_QUEUED;
 323                 lwkt_replymsg(msg, ECONNABORTED); /* YYY dangerous from IPI? */
 324                 goto again;
 325             }
 326             msg = TAILQ_NEXT(msg, ms_node);
 327         }
 328     } else {
 329         lwkt_send_ipiq(td->td_cpu, (ipifunc_t)lwkt_abortport_remote, port);
 330     }
 331 }
 332
 333 static
 334 void
 335 lwkt_abortport_remote(lwkt_port_t port)
 336 {
 337     _lwkt_abortport(port);
 338 }
 339
 340 void
 341 lwkt_abortport(lwkt_port_t port, lwkt_msg_t msg)
 342 {
 343     msg->ms_abortreq = 1;
 344     crit_enter();
 345     _lwkt_abortport(port);
 346     crit_exit();
 347 }
 348
 349 /*
 350  * lwkt_getport()
 351  *
 352  *      Retrieve the next message from the port's message queue, return NULL
 353  *      if no messages are pending.
 354  *
 355  *      The calling thread MUST own the port.
 356  */
 357 void *
 358 lwkt_getport(lwkt_port_t port)
 359 {
 360     lwkt_msg_t msg;
 361
 362     KKASSERT(port->mp_td == curthread);
 363
 364     crit_enter();
 365     if ((msg = TAILQ_FIRST(&port->mp_msgq)) != NULL) {
 366         TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 367         msg->ms_flags &= ~MSGF_QUEUED;
 368     }
 369     crit_exit();
 370     return(msg);
 371 }
 372
 373 /*
 374  * lwkt_waitport()
 375  *
 376  *      Retrieve the next message from the port's message queue, block until
 377  *      a message is ready.  This function never returns NULL.
 378  */
 379 void *
 380 lwkt_waitport(lwkt_port_t port)
 381 {
 382     lwkt_msg_t msg;
 383
 384     KKASSERT(port->mp_td == curthread);
 385
 386     crit_enter();
 387     if ((msg = TAILQ_FIRST(&port->mp_msgq)) == NULL) {
 388         port->mp_flags |= MSGPORTF_WAITING;
 389         do {
 390             lwkt_deschedule_self();
 391             lwkt_switch();
 392         } while ((msg = TAILQ_FIRST(&port->mp_msgq)) == NULL);
 393         port->mp_flags &= ~MSGPORTF_WAITING;
 394     }
 395     TAILQ_REMOVE(&port->mp_msgq, msg, ms_node);
 396     msg->ms_flags &= ~MSGF_QUEUED;
 397     crit_exit();
 398     return(msg);
 399 }
 400