Add a MSGF_NORESCHED feature for lwkt thread-based message ports. The
authorMatthew Dillon <dillon@dragonflybsd.org>
Tue, 9 Sep 2008 07:21:57 +0000 (07:21 +0000)
committerMatthew Dillon <dillon@dragonflybsd.org>
Tue, 9 Sep 2008 07:21:57 +0000 (07:21 +0000)
idea is to use it to allow certain async messages to be queued to higher
priority system threads and schedule those threads without forcing an
immediate reschedule.

The feature will be used by the new socket code to prevent cavitation
between a user process and system protocol thread when the user process
is write()ing a lot of data over the network.

sys/kern/lwkt_msgport.c
sys/kern/lwkt_thread.c
sys/sys/msgport.h
sys/sys/thread.h

index 224339c..7a7e383 100644 (file)
@@ -34,7 +34,7 @@
  * NOTE! This file may be compiled for userland libraries as well as for
  * the kernel.
  *
- * $DragonFly: src/sys/kern/lwkt_msgport.c,v 1.46 2008/05/18 20:57:56 nth Exp $
+ * $DragonFly: src/sys/kern/lwkt_msgport.c,v 1.47 2008/09/09 07:21:54 dillon Exp $
  */
 
 #include <sys/param.h>
@@ -229,6 +229,24 @@ _lwkt_initport(lwkt_port_t port,
     port->mp_replyport = rportfn;
 }
 
+/*
+ * Schedule the target thread.  If the message flags contains MSGF_NORESCHED
+ * we tell the scheduler not to reschedule if td is at a higher priority.
+ *
+ * This routine is called even if the thread is already scheduled so messages
+ * without NORESCHED will cause the target thread to be rescheduled even if
+ * prior messages did not.
+ */
+static __inline
+void
+_lwkt_schedule_msg(thread_t td, int flags)
+{
+       if (flags & MSGF_NORESCHED)
+           lwkt_schedule_noresched(td);
+       else
+           lwkt_schedule(td);
+}
+
 /*
  * lwkt_initport_thread()
  *
@@ -367,6 +385,7 @@ void
 lwkt_thread_replyport_remote(lwkt_msg_t msg)
 {
     lwkt_port_t port = msg->ms_reply_port;
+    int flags;
 
     /*
      * Chase any thread migration that occurs
@@ -384,14 +403,16 @@ lwkt_thread_replyport_remote(lwkt_msg_t msg)
     KKASSERT(msg->ms_flags & MSGF_INTRANSIT);
     msg->ms_flags &= ~MSGF_INTRANSIT;
 #endif
+    flags = msg->ms_flags;
     if (msg->ms_flags & MSGF_SYNC) {
-           msg->ms_flags |= MSGF_REPLY | MSGF_DONE;
+       cpu_sfence();
+       msg->ms_flags |= MSGF_REPLY | MSGF_DONE;
     } else {
-           TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
-           msg->ms_flags |= MSGF_REPLY | MSGF_DONE | MSGF_QUEUED;
+       TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
+       msg->ms_flags |= MSGF_REPLY | MSGF_DONE | MSGF_QUEUED;
     }
     if (port->mp_flags & MSGPORTF_WAITING)
-       lwkt_schedule(port->mpu_td);
+       _lwkt_schedule_msg(port->mpu_td, flags);
 }
 
 #endif
@@ -408,6 +429,8 @@ lwkt_thread_replyport_remote(lwkt_msg_t msg)
 void
 lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg)
 {
+    int flags;
+
     KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED|MSGF_INTRANSIT)) == 0);
 
     if (msg->ms_flags & MSGF_SYNC) {
@@ -421,9 +444,11 @@ lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg)
 #ifdef SMP
        if (port->mpu_td->td_gd == mycpu) {
 #endif
+           flags = msg->ms_flags;
+           cpu_sfence();
            msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
            if (port->mp_flags & MSGPORTF_WAITING)
-               lwkt_schedule(port->mpu_td);
+               _lwkt_schedule_msg(port->mpu_td, flags);
 #ifdef SMP
        } else {
 #ifdef INVARIANTS
@@ -448,7 +473,7 @@ lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg)
            TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
            msg->ms_flags |= MSGF_REPLY | MSGF_DONE | MSGF_QUEUED;
            if (port->mp_flags & MSGPORTF_WAITING)
-               lwkt_schedule(port->mpu_td);
+               _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
            crit_exit();
 #ifdef SMP
        } else {
@@ -500,7 +525,7 @@ lwkt_thread_putport_remote(lwkt_msg_t msg)
     TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
     msg->ms_flags |= MSGF_QUEUED;
     if (port->mp_flags & MSGPORTF_WAITING)
-       lwkt_schedule(port->mpu_td);
+       _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
 }
 
 #endif
@@ -519,7 +544,7 @@ lwkt_thread_putport(lwkt_port_t port, lwkt_msg_t msg)
        msg->ms_flags |= MSGF_QUEUED;
        TAILQ_INSERT_TAIL(&port->mp_msgq, msg, ms_node);
        if (port->mp_flags & MSGPORTF_WAITING)
-           lwkt_schedule(port->mpu_td);
+           _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
        crit_exit();
 #ifdef SMP
     } else {
index 71348f9..04e3537 100644 (file)
@@ -31,7 +31,7 @@
  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  * SUCH DAMAGE.
  * 
- * $DragonFly: src/sys/kern/lwkt_thread.c,v 1.117 2008/09/09 04:06:13 dillon Exp $
+ * $DragonFly: src/sys/kern/lwkt_thread.c,v 1.118 2008/09/09 07:21:54 dillon Exp $
  */
 
 /*
@@ -961,17 +961,23 @@ lwkt_yield(void)
  * It is possible for this routine to be called after a failed _enqueue
  * (due to the target thread migrating, sleeping, or otherwise blocked).
  * We have to check that the thread is actually on the run queue!
+ *
+ * reschedok is an optimized constant propagated from lwkt_schedule() or
+ * lwkt_schedule_noresched().  By default it is non-zero, causing a
+ * reschedule to be requested if the target thread has a higher priority.
+ * The port messaging code will set MSG_NORESCHED and cause reschedok to
+ * be 0, prevented undesired reschedules.
  */
 static __inline
 void
-_lwkt_schedule_post(globaldata_t gd, thread_t ntd, int cpri)
+_lwkt_schedule_post(globaldata_t gd, thread_t ntd, int cpri, int reschedok)
 {
     int mypri;
 
     if (ntd->td_flags & TDF_RUNQ) {
-       if (ntd->td_preemptable) {
+       if (ntd->td_preemptable && reschedok) {
            ntd->td_preemptable(ntd, cpri);     /* YYY +token */
-       } else {
+       } else if (reschedok) {
            /*
             * This is a little sticky.  Due to the passive release function
             * the LWKT priority can wiggle around for threads acting in
@@ -993,8 +999,9 @@ _lwkt_schedule_post(globaldata_t gd, thread_t ntd, int cpri)
     }
 }
 
+static __inline
 void
-lwkt_schedule(thread_t td)
+_lwkt_schedule(thread_t td, int reschedok)
 {
     globaldata_t mygd = mycpu;
 
@@ -1012,18 +1019,30 @@ lwkt_schedule(thread_t td)
 #ifdef SMP
        if (td->td_gd == mygd) {
            _lwkt_enqueue(td);
-           _lwkt_schedule_post(mygd, td, TDPRI_CRIT);
+           _lwkt_schedule_post(mygd, td, TDPRI_CRIT, reschedok);
        } else {
            lwkt_send_ipiq(td->td_gd, (ipifunc1_t)lwkt_schedule, td);
        }
 #else
        _lwkt_enqueue(td);
-       _lwkt_schedule_post(mygd, td, TDPRI_CRIT);
+       _lwkt_schedule_post(mygd, td, TDPRI_CRIT, reschedok);
 #endif
     }
     crit_exit_gd(mygd);
 }
 
+void
+lwkt_schedule(thread_t td)
+{
+    _lwkt_schedule(td, 1);
+}
+
+void
+lwkt_schedule_noresched(thread_t td)
+{
+    _lwkt_schedule(td, 0);
+}
+
 #ifdef SMP
 
 /*
index 5fd9c51..00b0738 100644 (file)
@@ -3,7 +3,7 @@
  *
  *     Implements LWKT messages and ports.
  * 
- * $DragonFly: src/sys/sys/msgport.h,v 1.28 2008/08/25 23:34:34 dillon Exp $
+ * $DragonFly: src/sys/sys/msgport.h,v 1.29 2008/09/09 07:21:57 dillon Exp $
  */
 
 #ifndef _SYS_MSGPORT_H_
@@ -98,6 +98,7 @@ typedef struct lwkt_msg {
 #define MSGF_QUEUED    0x0004          /* message has been queued sanitychk */
 #define MSGF_SYNC      0x0008          /* synchronous message operation */
 #define MSGF_INTRANSIT 0x0010          /* in-transit (IPI) */
+#define MSGF_NORESCHED 0x0020          /* do not reschedule target lwkt */
 #define MSGF_ABORTABLE 0x0080          /* message supports abort */
 
 #define MSGF_USER0     0x00010000
index 8a277dd..143c9bc 100644 (file)
@@ -7,7 +7,7 @@
  * Types which must already be defined when this header is included by
  * userland:   struct md_thread
  * 
- * $DragonFly: src/sys/sys/thread.h,v 1.95 2008/09/09 04:06:20 dillon Exp $
+ * $DragonFly: src/sys/sys/thread.h,v 1.96 2008/09/09 07:21:57 dillon Exp $
  */
 
 #ifndef _SYS_THREAD_H_
@@ -332,6 +332,7 @@ extern void lwkt_gdinit(struct globaldata *);
 extern void lwkt_switch(void);
 extern void lwkt_preempt(thread_t, int);
 extern void lwkt_schedule(thread_t);
+extern void lwkt_schedule_noresched(thread_t);
 extern void lwkt_schedule_self(thread_t);
 extern void lwkt_deschedule(thread_t);
 extern void lwkt_deschedule_self(thread_t);