Skip to content

Commit 5895eb4

Browse files
Jilleklauspost
andauthored
s2: Add AsyncFlush method: Complete the block without flushing (#927)
* s2: Add AsyncFlush method: Complete the block without flushing My use case is to transfer a large compressed S2 stream with a few changes very often. To get a small diff I want to end blocks at application decided points rather than at byte offsets. This allows me to remove the first byte without every single block changing. Flush() works for this, but it limits concurrency because it waits for the last block to be compressed rather than allowing that asynchronously. So I'd like to propose AsyncFlush, which flushes the buffer to a block, but doesn't flush the block to the io.Writer. There were actually a few places in the s2 code that also wanted to end the block, but didn't necessary want to flush to the writer. * Update s2/writer.go Co-authored-by: Klaus Post <[email protected]> --------- Co-authored-by: Klaus Post <[email protected]>
1 parent 4c49017 commit 5895eb4

File tree

1 file changed

+15
-6
lines changed

1 file changed

+15
-6
lines changed

s2/writer.go

+15-6
Original file line numberDiff line numberDiff line change
@@ -215,7 +215,7 @@ func (w *Writer) ReadFrom(r io.Reader) (n int64, err error) {
215215
return 0, err
216216
}
217217
if len(w.ibuf) > 0 {
218-
err := w.Flush()
218+
err := w.AsyncFlush()
219219
if err != nil {
220220
return 0, err
221221
}
@@ -225,7 +225,7 @@ func (w *Writer) ReadFrom(r io.Reader) (n int64, err error) {
225225
if err := w.EncodeBuffer(buf); err != nil {
226226
return 0, err
227227
}
228-
return int64(len(buf)), w.Flush()
228+
return int64(len(buf)), w.AsyncFlush()
229229
}
230230
for {
231231
inbuf := w.buffers.Get().([]byte)[:w.blockSize+obufHeaderLen]
@@ -354,7 +354,7 @@ func (w *Writer) EncodeBuffer(buf []byte) (err error) {
354354
}
355355
// Flush queued data first.
356356
if len(w.ibuf) > 0 {
357-
err := w.Flush()
357+
err := w.AsyncFlush()
358358
if err != nil {
359359
return err
360360
}
@@ -716,9 +716,9 @@ func (w *Writer) writeSync(p []byte) (nRet int, errRet error) {
716716
return nRet, nil
717717
}
718718

719-
// Flush flushes the Writer to its underlying io.Writer.
720-
// This does not apply padding.
721-
func (w *Writer) Flush() error {
719+
// AsyncFlush writes any buffered bytes to a block and starts compressing it.
720+
// It does not wait for the output has been written as Flush() does.
721+
func (w *Writer) AsyncFlush() error {
722722
if err := w.err(nil); err != nil {
723723
return err
724724
}
@@ -738,6 +738,15 @@ func (w *Writer) Flush() error {
738738
}
739739
}
740740
}
741+
return w.err(nil)
742+
}
743+
744+
// Flush flushes the Writer to its underlying io.Writer.
745+
// This does not apply padding.
746+
func (w *Writer) Flush() error {
747+
if err := w.AsyncFlush(); err != nil {
748+
return err
749+
}
741750
if w.output == nil {
742751
return w.err(nil)
743752
}

0 commit comments

Comments
 (0)