- Implement selwakeuppri() which allows raising the priority of a

  thread being waken up.  The thread waken up can run at a priority as
  high as after tsleep().

- Replace selwakeup()s with selwakeuppri()s and pass appropriate
  priorities.

- Add cv_broadcastpri() which raises the priority of the broadcast
  threads.  Used by selwakeuppri() if collision occurs.

Not objected in:	-arch, -current

11 files changed, 45 insertions, 17 deletions

diff --git a/sys/kern/kern_condvar.c b/sys/kern/kern_condvar.c
index 3344123..d95d9e2 100644
--- a/sys/kern/kern_condvar.c
+++ b/sys/kern/kern_condvar.c
@@ -525,14 +525,21 @@ cv_signal(struct cv *cvp)
  * Should be called with the same mutex as was passed to cv_wait held.
  */
 void
-cv_broadcast(struct cv *cvp)
+cv_broadcastpri(struct cv *cvp, int pri)
 {
+	struct thread	*td;
 
 	KASSERT(cvp != NULL, ("%s: cvp NULL", __func__));
 	mtx_lock_spin(&sched_lock);
 	CV_SIGNAL_VALIDATE(cvp);
-	while (!TAILQ_EMPTY(&cvp->cv_waitq))
+	while (!TAILQ_EMPTY(&cvp->cv_waitq)) {
+		if (pri >= PRI_MIN && pri <= PRI_MAX) {
+			td = TAILQ_FIRST(&cvp->cv_waitq);
+			if (td->td_priority > pri)
+				td->td_priority = pri;
+		}
 		cv_wakeup(cvp);
+	}
 	mtx_unlock_spin(&sched_lock);
 }
 
diff --git a/sys/kern/kern_event.c b/sys/kern/kern_event.c
index 63a914f..a0229a7 100644
--- a/sys/kern/kern_event.c
+++ b/sys/kern/kern_event.c
@@ -925,7 +925,7 @@ kqueue_wakeup(struct kqueue *kq)
 	}
 	if (kq->kq_state & KQ_SEL) {
 		kq->kq_state &= ~KQ_SEL;
-		selwakeup(&kq->kq_sel);
+		selwakeuppri(&kq->kq_sel, PSOCK);
 	}
 	KNOTE(&kq->kq_sel.si_note, 0);
 }
diff --git a/sys/kern/subr_log.c b/sys/kern/subr_log.c
index 031d081..b2ece2d 100644
--- a/sys/kern/subr_log.c
+++ b/sys/kern/subr_log.c
@@ -192,7 +192,7 @@ logtimeout(void *arg)
 		return;
 	}
 	msgbuftrigger = 0;
-	selwakeup(&logsoftc.sc_selp);
+	selwakeuppri(&logsoftc.sc_selp, LOG_RDPRI);
 	if ((logsoftc.sc_state & LOG_ASYNC) && logsoftc.sc_sigio != NULL)
 		pgsigio(&logsoftc.sc_sigio, SIGIO, 0);
 	if (logsoftc.sc_state & LOG_RDWAIT) {
diff --git a/sys/kern/sys_generic.c b/sys/kern/sys_generic.c
index 221ec00..881be56 100644
--- a/sys/kern/sys_generic.c
+++ b/sys/kern/sys_generic.c
@@ -82,6 +82,7 @@ static int	dofileread(struct thread *, struct file *, int, void *,
 		    size_t, off_t, int);
 static int	dofilewrite(struct thread *, struct file *, int,
 		    const void *, size_t, off_t, int);
+static void	doselwakeup(struct selinfo *, int);
 
 /*
  * Read system call.
@@ -1162,12 +1163,30 @@ selrecord(selector, sip)
 	mtx_unlock(&sellock);
 }
 
+/* Wake up a selecting thread. */
+void
+selwakeup(sip)
+	struct selinfo *sip;
+{
+	doselwakeup(sip, -1);
+}
+
+/* Wake up a selecting thread, and set its priority. */
+void
+selwakeuppri(sip, pri)
+	struct selinfo *sip;
+	int pri;
+{
+	doselwakeup(sip, pri);
+}
+
 /*
  * Do a wakeup when a selectable event occurs.
  */
-void
-selwakeup(sip)
+static void
+doselwakeup(sip, pri)
 	struct selinfo *sip;
+	int pri;
 {
 	struct thread *td;
 
@@ -1176,7 +1195,7 @@ selwakeup(sip)
 	if ((sip->si_flags & SI_COLL) != 0) {
 		nselcoll++;
 		sip->si_flags &= ~SI_COLL;
-		cv_broadcast(&selwait);
+		cv_broadcastpri(&selwait, pri);
 	}
 	if (td == NULL) {
 		mtx_unlock(&sellock);
@@ -1188,6 +1207,8 @@ selwakeup(sip)
 	if (td->td_wchan == &selwait) {
 		cv_waitq_remove(td);
 		TD_CLR_SLEEPING(td);
+		if (pri >= PRI_MIN && pri <= PRI_MAX && td->td_priority > pri)
+			td->td_priority = pri;
 		setrunnable(td);
 	} else
 		td->td_flags &= ~TDF_SELECT;
diff --git a/sys/kern/sys_pipe.c b/sys/kern/sys_pipe.c
index 3208e5c..7b4667b 100644
--- a/sys/kern/sys_pipe.c
+++ b/sys/kern/sys_pipe.c
@@ -447,7 +447,7 @@ pipeselwakeup(cpipe)
 
 	if (cpipe->pipe_state & PIPE_SEL) {
 		cpipe->pipe_state &= ~PIPE_SEL;
-		selwakeup(&cpipe->pipe_sel);
+		selwakeuppri(&cpipe->pipe_sel, PSOCK);
 	}
 	if ((cpipe->pipe_state & PIPE_ASYNC) && cpipe->pipe_sigio)
 		pgsigio(&cpipe->pipe_sigio, SIGIO, 0);
diff --git a/sys/kern/tty.c b/sys/kern/tty.c
index 4bebd65..f34858c 100644
--- a/sys/kern/tty.c
+++ b/sys/kern/tty.c
@@ -2271,7 +2271,7 @@ ttwakeup(struct tty *tp)
 {
 
 	if (SEL_WAITING(&tp->t_rsel))
-		selwakeup(&tp->t_rsel);
+		selwakeuppri(&tp->t_rsel, TTIPRI);
 	if (ISSET(tp->t_state, TS_ASYNC) && tp->t_sigio != NULL)
 		pgsigio(&tp->t_sigio, SIGIO, (tp->t_session != NULL));
 	wakeup(TSA_HUP_OR_INPUT(tp));
@@ -2286,7 +2286,7 @@ ttwwakeup(struct tty *tp)
 {
 
 	if (SEL_WAITING(&tp->t_wsel) && tp->t_outq.c_cc <= tp->t_olowat)
-		selwakeup(&tp->t_wsel);
+		selwakeuppri(&tp->t_wsel, TTOPRI);
 	if (ISSET(tp->t_state, TS_ASYNC) && tp->t_sigio != NULL)
 		pgsigio(&tp->t_sigio, SIGIO, (tp->t_session != NULL));
 	if (ISSET(tp->t_state, TS_BUSY | TS_SO_OCOMPLETE) ==
diff --git a/sys/kern/tty_pty.c b/sys/kern/tty_pty.c
index 2de3a04..874f0ba 100644
--- a/sys/kern/tty_pty.c
+++ b/sys/kern/tty_pty.c
@@ -326,11 +326,11 @@ ptcwakeup(tp, flag)
 	struct pt_ioctl *pti = tp->t_dev->si_drv1;
 
 	if (flag & FREAD) {
-		selwakeup(&pti->pt_selr);
+		selwakeuppri(&pti->pt_selr, TTIPRI);
 		wakeup(TSA_PTC_READ(tp));
 	}
 	if (flag & FWRITE) {
-		selwakeup(&pti->pt_selw);
+		selwakeuppri(&pti->pt_selw, TTOPRI);
 		wakeup(TSA_PTC_WRITE(tp));
 	}
 }
diff --git a/sys/kern/uipc_sockbuf.c b/sys/kern/uipc_sockbuf.c
index d64198d..c4a4c09 100644
--- a/sys/kern/uipc_sockbuf.c
+++ b/sys/kern/uipc_sockbuf.c
@@ -308,7 +308,7 @@ sowakeup(so, sb)
 	register struct sockbuf *sb;
 {
 
-	selwakeup(&sb->sb_sel);
+	selwakeuppri(&sb->sb_sel, PSOCK);
 	sb->sb_flags &= ~SB_SEL;
 	if (sb->sb_flags & SB_WAIT) {
 		sb->sb_flags &= ~SB_WAIT;
diff --git a/sys/kern/uipc_socket.c b/sys/kern/uipc_socket.c
index 4d4eb63..8e5dec2 100644
--- a/sys/kern/uipc_socket.c
+++ b/sys/kern/uipc_socket.c
@@ -1744,7 +1744,7 @@ sohasoutofband(so)
 {
 	if (so->so_sigio != NULL)
 		pgsigio(&so->so_sigio, SIGURG, 0);
-	selwakeup(&so->so_rcv.sb_sel);
+	selwakeuppri(&so->so_rcv.sb_sel, PSOCK);
 }
 
 int
diff --git a/sys/kern/uipc_socket2.c b/sys/kern/uipc_socket2.c
index d64198d..c4a4c09 100644
--- a/sys/kern/uipc_socket2.c
+++ b/sys/kern/uipc_socket2.c
@@ -308,7 +308,7 @@ sowakeup(so, sb)
 	register struct sockbuf *sb;
 {
 
-	selwakeup(&sb->sb_sel);
+	selwakeuppri(&sb->sb_sel, PSOCK);
 	sb->sb_flags &= ~SB_SEL;
 	if (sb->sb_flags & SB_WAIT) {
 		sb->sb_flags &= ~SB_WAIT;
diff --git a/sys/kern/vfs_subr.c b/sys/kern/vfs_subr.c
index dc9934b..8a49c13 100644
--- a/sys/kern/vfs_subr.c
+++ b/sys/kern/vfs_subr.c
@@ -3375,7 +3375,7 @@ vn_pollevent(vp, events)
 		 */
 		vp->v_pollinfo->vpi_events = 0;	/* &= ~events ??? */
 		vp->v_pollinfo->vpi_revents |= events;
-		selwakeup(&vp->v_pollinfo->vpi_selinfo);
+		selwakeuppri(&vp->v_pollinfo->vpi_selinfo, PRIBIO);
 	}
 	mtx_unlock(&vp->v_pollinfo->vpi_lock);
 }
@@ -3394,7 +3394,7 @@ vn_pollgone(vp)
 	VN_KNOTE(vp, NOTE_REVOKE);
 	if (vp->v_pollinfo->vpi_events) {
 		vp->v_pollinfo->vpi_events = 0;
-		selwakeup(&vp->v_pollinfo->vpi_selinfo);
+		selwakeuppri(&vp->v_pollinfo->vpi_selinfo, PRIBIO);
 	}
 	mtx_unlock(&vp->v_pollinfo->vpi_lock);
 }