diff --git a/internal/multiplex/session.go b/internal/multiplex/session.go index ecb2dbf..868ebab 100644 --- a/internal/multiplex/session.go +++ b/internal/multiplex/session.go @@ -165,7 +165,9 @@ func (sesh *Session) closeStream(s *Stream, active bool) error { Payload: pad, } - s.allocIdempot.Do(func() { s.obfsBuf = make([]byte, s.session.SendBufferSize) }) + if s.obfsBuf == nil { + s.obfsBuf = make([]byte, s.session.SendBufferSize) + } i, err := s.session.Obfs(f, s.obfsBuf) if err != nil { return err diff --git a/internal/multiplex/stream.go b/internal/multiplex/stream.go index dbccae0..3dfb8a2 100644 --- a/internal/multiplex/stream.go +++ b/internal/multiplex/stream.go @@ -23,7 +23,7 @@ type Stream struct { // atomic nextSendSeq uint64 - writingM sync.RWMutex + writingM sync.Mutex // atomic closed uint32 @@ -92,14 +92,15 @@ func (s *Stream) Write(in []byte) (n int, err error) { // to be sent before the data frame and cause loss of packet. //log.Tracef("attempting to write %v bytes to stream %v",len(in),s.id) // todo: forbid concurrent write - s.writingM.RLock() - defer s.writingM.RUnlock() + s.writingM.Lock() + defer s.writingM.Unlock() if s.isClosed() { return 0, ErrBrokenStream } - s.allocIdempot.Do(func() { s.obfsBuf = make([]byte, s.session.SendBufferSize) }) - + if s.obfsBuf == nil { + s.obfsBuf = make([]byte, s.session.SendBufferSize) + } for n < len(in) { var framePayload []byte if len(in)-n <= s.session.maxStreamUnitWrite {