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