xl: Simplify blockingWriter and its usage. (#1373)

This removes odd races since we don't need to
track errors and avoids locking. All we need
is a Wait() and Done() waitgroup.
This commit is contained in:
Harshavardhana 2016-04-25 00:27:04 -07:00 committed by Harshavardhana
parent 8bce699dae
commit 55032ffdf9
2 changed files with 6 additions and 31 deletions

View File

@ -25,43 +25,26 @@ import (
type blockingWriteCloser struct { type blockingWriteCloser struct {
writer io.WriteCloser // Embedded writer. writer io.WriteCloser // Embedded writer.
release *sync.WaitGroup // Waitgroup for atomicity. release *sync.WaitGroup // Waitgroup for atomicity.
mutex *sync.Mutex // Mutex for thread safety.
err error err error
} }
// Write to the underlying writer. // Write to the underlying writer.
func (b *blockingWriteCloser) Write(data []byte) (int, error) { func (b *blockingWriteCloser) Write(data []byte) (int, error) {
n, err := b.writer.Write(data) return b.writer.Write(data)
if err != nil {
b.mutex.Lock()
b.err = err
b.mutex.Unlock()
}
return n, b.err
} }
// Close blocks until another goroutine calls Release(error). Returns // Close blocks until another goroutine calls Release(error). Returns
// error code if either writer fails or Release is called with an error. // error code if either writer fails or Release is called with an error.
func (b *blockingWriteCloser) Close() error { func (b *blockingWriteCloser) Close() error {
err := b.writer.Close() err := b.writer.Close()
if err != nil {
b.mutex.Lock()
b.err = err
b.mutex.Unlock()
}
b.release.Wait() b.release.Wait()
return b.err return err
} }
// Release the Close, causing it to unblock. Only call this // Release the Close, causing it to unblock. Only call this
// once. Calling it multiple times results in a panic. // once. Calling it multiple times results in a panic.
func (b *blockingWriteCloser) Release(err error) { func (b *blockingWriteCloser) Release() {
b.release.Done() b.release.Done()
if err != nil {
b.mutex.Lock()
b.err = err
b.mutex.Unlock()
}
return return
} }
@ -74,7 +57,6 @@ func newBlockingWriteCloser(writer io.WriteCloser) *blockingWriteCloser {
wg.Add(1) wg.Add(1)
return &blockingWriteCloser{ return &blockingWriteCloser{
writer: writer, writer: writer,
mutex: &sync.Mutex{},
release: wg, release: wg,
} }
} }

View File

@ -129,6 +129,9 @@ func (xl XL) getFileQuorumVersionMap(volume, path string) map[int]int64 {
// WriteErasure reads predefined blocks, encodes them and writes to // WriteErasure reads predefined blocks, encodes them and writes to
// configured storage disks. // configured storage disks.
func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *blockingWriteCloser) { func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *blockingWriteCloser) {
// Release the block writer upon function return.
defer bwriter.Release()
// Get available quorum for existing file path. // Get available quorum for existing file path.
_, higherVersion := xl.getQuorumDisks(volume, path) _, higherVersion := xl.getQuorumDisks(volume, path)
// Increment to have next higher version. // Increment to have next higher version.
@ -163,7 +166,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove previous temp writers for any failure. // Remove previous temp writers for any failure.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(errWriteQuorum) reader.CloseWithError(errWriteQuorum)
bwriter.Release(errWriteQuorum)
return return
} }
@ -186,7 +188,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove previous temp writers for any failure. // Remove previous temp writers for any failure.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(errWriteQuorum) reader.CloseWithError(errWriteQuorum)
bwriter.Release(errWriteQuorum)
return return
} }
@ -207,7 +208,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove all temp writers. // Remove all temp writers.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(err) reader.CloseWithError(err)
bwriter.Release(err)
return return
} }
} }
@ -223,7 +223,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove all temp writers. // Remove all temp writers.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(err) reader.CloseWithError(err)
bwriter.Release(err)
return return
} }
@ -233,7 +232,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove all temp writers upon error. // Remove all temp writers upon error.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(err) reader.CloseWithError(err)
bwriter.Release(err)
return return
} }
@ -248,7 +246,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove all temp writers upon error. // Remove all temp writers upon error.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(err) reader.CloseWithError(err)
bwriter.Release(err)
return return
} }
if sha512Writers[index] != nil { if sha512Writers[index] != nil {
@ -298,7 +295,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
// Remove temporary files. // Remove temporary files.
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...) xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
reader.CloseWithError(err) reader.CloseWithError(err)
bwriter.Release(err)
return return
} }
} }
@ -322,9 +318,6 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, bwriter *b
metadataWriters[index].Close() metadataWriters[index].Close()
} }
// Release the blocking writer.
bwriter.Release(nil)
// Close the pipe reader and return. // Close the pipe reader and return.
reader.Close() reader.Close()
return return