http/spdy: fix data race in header decompression.

flate's reader greedily reads from the shared io.Reader in Framer. This leads to a data race on Framer.r. Fix this by providing a corkedReader to zlib.NewReaderDict(). We uncork the reader and allow it to read the number of bytes in the compressed payload.

Fixes #1884.

R=bradfitz, rsc, go.peter.90
CC=golang-dev
https://golang.org/cl/4530089

diff --git a/src/pkg/http/spdy/framer.go b/src/pkg/http/spdy/framer.go
index 0ad2989d27f0251691fde532543880e09d11fcaa..8850c71083867309364bfd3407868afdfbfcdb70 100644 (file)
--- a/src/pkg/http/spdy/framer.go
+++ b/src/pkg/http/spdy/framer.go
@@ -44,6 +44,24 @@ func (e FramerError) String() string {
        return "Error(" + strconv.Itoa(int(e)) + ")"
 }
 
+type corkedReader struct {
+       r  io.Reader
+       ch chan int
+       n  int
+}
+
+func (cr *corkedReader) Read(p []byte) (int, os.Error) {
+       if cr.n == 0 {
+               cr.n = <-cr.ch
+       }
+       if len(p) > cr.n {
+               p = p[:cr.n]
+       }
+       n, err := cr.r.Read(p)
+       cr.n -= n
+       return n, err
+}
+
 // Framer handles serializing/deserializing SPDY frames, including compressing/
 // decompressing payloads.
 type Framer struct {
@@ -52,6 +70,7 @@ type Framer struct {
        headerBuf                 *bytes.Buffer
        headerCompressor          *zlib.Writer
        r                         io.Reader
+       headerReader              corkedReader
        headerDecompressor        io.ReadCloser
 }
 
@@ -74,11 +93,13 @@ func NewFramer(w io.Writer, r io.Reader) (*Framer, os.Error) {
        return framer, nil
 }
 
-func (f *Framer) initHeaderDecompression() os.Error {
+func (f *Framer) uncorkHeaderDecompressor(payloadSize int) os.Error {
        if f.headerDecompressor != nil {
+               f.headerReader.ch <- payloadSize
                return nil
        }
-       decompressor, err := zlib.NewReaderDict(f.r, []byte(HeaderDictionary))
+       f.headerReader = corkedReader{r: f.r, ch: make(chan int, 1), n: payloadSize}
+       decompressor, err := zlib.NewReaderDict(&f.headerReader, []byte(HeaderDictionary))
        if err != nil {
                return err
        }
@@ -171,7 +192,7 @@ func (f *Framer) readSynStreamFrame(h ControlFrameHeader, frame *SynStreamFrame)
 
        reader := f.r
        if !f.headerCompressionDisabled {
-               f.initHeaderDecompression()
+               f.uncorkHeaderDecompressor(int(h.length - 10))
                reader = f.headerDecompressor
        }
 
@@ -194,7 +215,7 @@ func (f *Framer) readSynReplyFrame(h ControlFrameHeader, frame *SynReplyFrame) o
        }
        reader := f.r
        if !f.headerCompressionDisabled {
-               f.initHeaderDecompression()
+               f.uncorkHeaderDecompressor(int(h.length - 6))
                reader = f.headerDecompressor
        }
        frame.Headers, err = parseHeaderValueBlock(reader)
@@ -216,7 +237,7 @@ func (f *Framer) readHeadersFrame(h ControlFrameHeader, frame *HeadersFrame) os.
        }
        reader := f.r
        if !f.headerCompressionDisabled {
-               f.initHeaderDecompression()
+               f.uncorkHeaderDecompressor(int(h.length - 6))
                reader = f.headerDecompressor
        }
        frame.Headers, err = parseHeaderValueBlock(reader)

diff --git a/src/pkg/http/spdy/framer_test.go b/src/pkg/http/spdy/framer_test.go
index 192d688fb65332cf97e70f89e43a73158b6c75c6..9100e1ea8978a7da60a63f41c260d4b9a9cb1940 100644 (file)
--- a/src/pkg/http/spdy/framer_test.go
+++ b/src/pkg/http/spdy/framer_test.go
@@ -371,12 +371,6 @@ func TestCreateParseDataFrame(t *testing.T) {
 }
 
 func TestCompressionContextAcrossFrames(t *testing.T) {
-       {
-               // TODO(willchan,bradfitz): test is temporarily disabled
-               t.Logf("test temporarily disabled; http://code.google.com/p/go/issues/detail?id=1884")
-               return
-       }
-
        buffer := new(bytes.Buffer)
        framer, err := NewFramer(buffer, buffer)
        if err != nil {
@@ -430,12 +424,6 @@ func TestCompressionContextAcrossFrames(t *testing.T) {
 }
 
 func TestMultipleSPDYFrames(t *testing.T) {
-       {
-               // TODO(willchan,bradfitz): test is temporarily disabled
-               t.Logf("test temporarily disabled; http://code.google.com/p/go/issues/detail?id=1884")
-               return
-       }
-
        // Initialize the framers.
        pr1, pw1 := io.Pipe()
        pr2, pw2 := io.Pipe()