update

session.go

@@ -254,8 +254,8 @@ func (s *Session) PollWait(revents []*Stream, wevents []*Stream) (int, int, erro
 	}
 }
 
-// streams notify session events
-func (s *Session) notifyPoller(stream *Stream) {
+// streams notify session readable events
+func (s *Session) notifyPollIn(stream *Stream) {
 	s.pollEventsLock.Lock()
 	s.pollInEvents[stream.id] = stream
 	s.pollEventsLock.Unlock()
@@ -266,6 +266,18 @@ func (s *Session) notifyPoller(stream *Stream) {
 	}
 }
 
+// streams notify session writable eevents
+func (s *Session) notifyPollOut(stream *Stream) {
+	s.pollEventsLock.Lock()
+	s.pollOutEvents[stream.id] = stream
+	s.pollEventsLock.Unlock()
+
+	select {
+	case s.chPollNotify <- struct{}{}:
+	default:
+	}
+}
+
 // notifyBucket notifies recvLoop that bucket is available
 func (s *Session) notifyBucket() {
 	select {

stream.go

@@ -511,7 +511,7 @@ func (s *Stream) pushBytes(buf []byte) (written int, err error) {
 	s.heads = append(s.heads, buf)
 	// Edge-Trigger
 	if len(s.buffers) == 1 {
-		s.sess.notifyPoller(s)
+		s.sess.notifyPollIn(s)
 	}
 	s.bufferLock.Unlock()
 	return
@@ -547,7 +547,7 @@ func (s *Stream) update(consumed uint32, window uint32) {
 	default:
 	}
 
-	s.sess.notifyPoller(s)
+	s.sess.notifyPollOut(s)
 }
 
 // mark this stream has been closed in protocol