Imported Upstream version 2011.01.12upstream/2011.01.12

1 files changed, 114 insertions, 34 deletions

diff --git a/src/pkg/compress/flate/deflate.go b/src/pkg/compress/flate/deflate.go
index 79952e713..591b35c44 100644
--- a/src/pkg/compress/flate/deflate.go
+++ b/src/pkg/compress/flate/deflate.go
@@ -53,19 +53,19 @@ type compressionLevel struct {
 }
 
 var levels = []compressionLevel{
-	compressionLevel{}, // 0
+	{}, // 0
 	// For levels 1-3 we don't bother trying with lazy matches
-	compressionLevel{3, 0, 8, 4, 4},
-	compressionLevel{3, 0, 16, 8, 5},
-	compressionLevel{3, 0, 32, 32, 6},
+	{3, 0, 8, 4, 4},
+	{3, 0, 16, 8, 5},
+	{3, 0, 32, 32, 6},
 	// Levels 4-9 use increasingly more lazy matching
 	// and increasingly stringent conditions for "good enough".
-	compressionLevel{4, 4, 16, 16, math.MaxInt32},
-	compressionLevel{8, 16, 32, 32, math.MaxInt32},
-	compressionLevel{8, 16, 128, 128, math.MaxInt32},
-	compressionLevel{8, 32, 128, 256, math.MaxInt32},
-	compressionLevel{32, 128, 258, 1024, math.MaxInt32},
-	compressionLevel{32, 258, 258, 4096, math.MaxInt32},
+	{4, 4, 16, 16, math.MaxInt32},
+	{8, 16, 32, 32, math.MaxInt32},
+	{8, 16, 128, 128, math.MaxInt32},
+	{8, 32, 128, 256, math.MaxInt32},
+	{32, 128, 258, 1024, math.MaxInt32},
+	{32, 258, 258, 4096, math.MaxInt32},
 }
 
 func (sw *syncPipeWriter) Close() os.Error {
@@ -89,6 +89,10 @@ type compressor struct {
 	// (1 << logWindowSize) - 1.
 	windowMask int
 
+	eof      bool // has eof been reached on input?
+	sync     bool // writer wants to flush
+	syncChan chan os.Error
+
 	// hashHead[hashValue] contains the largest inputIndex with the specified hash value
 	hashHead []int
 
@@ -124,6 +128,9 @@ func (d *compressor) flush() os.Error {
 }
 
 func (d *compressor) fillWindow(index int) (int, os.Error) {
+	if d.sync {
+		return index, nil
+	}
 	wSize := d.windowMask + 1
 	if index >= wSize+wSize-(minMatchLength+maxMatchLength) {
 		// shift the window by wSize
@@ -142,12 +149,14 @@ func (d *compressor) fillWindow(index int) (int, os.Error) {
 			d.hashPrev[i] = max(h-wSize, -1)
 		}
 	}
-	var count int
-	var err os.Error
-	count, err = io.ReadAtLeast(d.r, d.window[d.windowEnd:], 1)
+	count, err := d.r.Read(d.window[d.windowEnd:])
 	d.windowEnd += count
+	if count == 0 && err == nil {
+		d.sync = true
+	}
 	if err == os.EOF {
-		return index, nil
+		d.eof = true
+		err = nil
 	}
 	return index, err
 }
@@ -227,10 +236,17 @@ func (d *compressor) storedDeflate() os.Error {
 	buf := make([]byte, maxStoreBlockSize)
 	for {
 		n, err := d.r.Read(buf)
-		if n > 0 {
+		if n == 0 && err == nil {
+			d.sync = true
+		}
+		if n > 0 || d.sync {
 			if err := d.writeStoredBlock(buf[0:n]); err != nil {
 				return err
 			}
+			if d.sync {
+				d.syncChan <- nil
+				d.sync = false
+			}
 		}
 		if err != nil {
 			if err == os.EOF {
@@ -275,6 +291,7 @@ func (d *compressor) doDeflate() (err os.Error) {
 		hash = int(d.window[index])<<hashShift + int(d.window[index+1])
 	}
 	chainHead := -1
+Loop:
 	for {
 		if index > windowEnd {
 			panic("index > windowEnd")
@@ -291,7 +308,31 @@ func (d *compressor) doDeflate() (err os.Error) {
 			maxInsertIndex = windowEnd - (minMatchLength - 1)
 			lookahead = windowEnd - index
 			if lookahead == 0 {
-				break
+				// Flush current output block if any.
+				if byteAvailable {
+					// There is still one pending token that needs to be flushed
+					tokens[ti] = literalToken(uint32(d.window[index-1]) & 0xFF)
+					ti++
+					byteAvailable = false
+				}
+				if ti > 0 {
+					if err = d.writeBlock(tokens[0:ti], index, false); err != nil {
+						return
+					}
+					ti = 0
+				}
+				if d.sync {
+					d.w.writeStoredHeader(0, false)
+					d.w.flush()
+					d.syncChan <- d.w.err
+					d.sync = false
+				}
+
+				// If this was only a sync (not at EOF) keep going.
+				if !d.eof {
+					continue
+				}
+				break Loop
 			}
 		}
 		if index < maxInsertIndex {
@@ -383,23 +424,11 @@ func (d *compressor) doDeflate() (err os.Error) {
 				byteAvailable = true
 			}
 		}
-
-	}
-	if byteAvailable {
-		// There is still one pending token that needs to be flushed
-		tokens[ti] = literalToken(uint32(d.window[index-1]) & 0xFF)
-		ti++
-	}
-
-	if ti > 0 {
-		if err = d.writeBlock(tokens[0:ti], index, false); err != nil {
-			return
-		}
 	}
 	return
 }
 
-func (d *compressor) compressor(r io.Reader, w io.Writer, level int, logWindowSize uint) (err os.Error) {
+func (d *compressor) compress(r io.Reader, w io.Writer, level int, logWindowSize uint) (err os.Error) {
 	d.r = r
 	d.w = newHuffmanBitWriter(w)
 	d.level = level
@@ -417,6 +446,10 @@ func (d *compressor) compressor(r io.Reader, w io.Writer, level int, logWindowSi
 		return WrongValueError{"level", 0, 9, int32(level)}
 	}
 
+	if d.sync {
+		d.syncChan <- err
+		d.sync = false
+	}
 	if err != nil {
 		return err
 	}
@@ -426,16 +459,63 @@ func (d *compressor) compressor(r io.Reader, w io.Writer, level int, logWindowSi
 	return d.flush()
 }
 
-func newCompressor(w io.Writer, level int, logWindowSize uint) io.WriteCloser {
+// NewWriter returns a new Writer compressing
+// data at the given level.  Following zlib, levels
+// range from 1 (BestSpeed) to 9 (BestCompression);
+// higher levels typically run slower but compress more.
+// Level 0 (NoCompression) does not attempt any
+// compression; it only adds the necessary DEFLATE framing.
+func NewWriter(w io.Writer, level int) *Writer {
+	const logWindowSize = logMaxOffsetSize
 	var d compressor
+	d.syncChan = make(chan os.Error, 1)
 	pr, pw := syncPipe()
 	go func() {
-		err := d.compressor(pr, w, level, logWindowSize)
+		err := d.compress(pr, w, level, logWindowSize)
 		pr.CloseWithError(err)
 	}()
-	return pw
+	return &Writer{pw, &d}
+}
+
+// A Writer takes data written to it and writes the compressed
+// form of that data to an underlying writer (see NewWriter).
+type Writer struct {
+	w *syncPipeWriter
+	d *compressor
+}
+
+// Write writes data to w, which will eventually write the
+// compressed form of data to its underlying writer.
+func (w *Writer) Write(data []byte) (n int, err os.Error) {
+	if len(data) == 0 {
+		// no point, and nil interferes with sync
+		return
+	}
+	return w.w.Write(data)
+}
+
+// Flush flushes any pending compressed data to the underlying writer.
+// It is useful mainly in compressed network protocols, to ensure that
+// a remote reader has enough data to reconstruct a packet.
+// Flush does not return until the data has been written.
+// If the underlying writer returns an error, Flush returns that error.
+//
+// In the terminology of the zlib library, Flush is equivalent to Z_SYNC_FLUSH.
+func (w *Writer) Flush() os.Error {
+	// For more about flushing:
+	// http://www.bolet.org/~pornin/deflate-flush.html
+	if w.d.sync {
+		panic("compress/flate: double Flush")
+	}
+	_, err := w.w.Write(nil)
+	err1 := <-w.d.syncChan
+	if err == nil {
+		err = err1
+	}
+	return err
 }
 
-func NewWriter(w io.Writer, level int) io.WriteCloser {
-	return newCompressor(w, level, logMaxOffsetSize)
+// Close flushes and closes the writer.
+func (w *Writer) Close() os.Error {
+	return w.w.Close()
 }