From b45c8940691522e0ad068f93dc91c5a2e72e2944 Mon Sep 17 00:00:00 2001 From: xtaci Date: Sun, 24 Mar 2019 18:40:36 +0800 Subject: [PATCH] fan in writeFrameInternal --- stream.go | 55 ++++++++++++++----------------------------------------- 1 file changed, 14 insertions(+), 41 deletions(-) diff --git a/stream.go b/stream.go index 2a2b82f..ce36f41 100644 --- a/stream.go +++ b/stream.go @@ -98,34 +98,24 @@ func (s *Stream) Write(b []byte) (n int, err error) { default: } - frames := s.split(b, cmdPSH, s.id) + // frame split and transmit sent := 0 - for k := range frames { - req := writeRequest{ - frame: frames[k], - result: make(chan writeResult, 1), + frame := newFrame(cmdPSH, s.id) + bts := b + for len(bts) > 0 { + sz := len(bts) + if sz > s.frameSize { + sz = s.frameSize } - - select { - case s.sess.writes <- req: - case <-s.die: - return sent, errors.New(errBrokenPipe) - case <-deadline: - return sent, errTimeout - } - - select { - case result := <-req.result: - sent += result.n - if result.err != nil { - return sent, result.err - } - case <-s.die: - return sent, errors.New(errBrokenPipe) - case <-deadline: - return sent, errTimeout + frame.data = bts[:sz] + bts = bts[sz:] + n, err := s.sess.writeFrameInternal(frame, deadline) + sent += n + if err != nil { + return sent, err } } + return sent, nil } @@ -229,23 +219,6 @@ func (s *Stream) recycleTokens() (n int) { return } -// split large byte buffer into smaller frames, reference only -func (s *Stream) split(bts []byte, cmd byte, sid uint32) []Frame { - frames := make([]Frame, 0, len(bts)/s.frameSize+1) - for len(bts) > s.frameSize { - frame := newFrame(cmd, sid) - frame.data = bts[:s.frameSize] - bts = bts[s.frameSize:] - frames = append(frames, frame) - } - if len(bts) > 0 { - frame := newFrame(cmd, sid) - frame.data = bts - frames = append(frames, frame) - } - return frames -} - // notify read event func (s *Stream) notifyReadEvent() { select {