maxmustermann/minio
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

fix: improve NewObjectReader implementation for careful cleanup usage (#12199)

Browse source 
cleanup functions should never be cleaned before the reader is
instantiated, this type of design leads to situations where order
of lockers and places for them to use becomes confusing.

Allow WithCleanupFuncs() if the caller wishes to add cleanupFns
to be run upon close() or an error during initialization of the
reader.

Also make sure streams are closed before we unlock the resources,
this allows for ordered cleanup of resources.
This commit is contained in:
Harshavardhana 2021-04-30 18:37:58 -07:00 committed by GitHub
parent 3524c00090
commit 0d3ddf7286
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 44 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/bucket-lifecycle.go
View file

@ -325,6 +325,7 @@ func getTransitionedObjectReader(ctx context.Context, bucket, object string, rs
if err != nil {
return nil, fmt.Errorf("transition storage class not configured")
}
fn, off, length, err := NewGetObjectReader(rs, oi, opts)
if err != nil {
return nil, ErrorRespToObjectError(err, bucket, object)

6
cmd/disk-cache-backend.go
View file

@ -936,16 +936,16 @@ func (c *diskCache) Get(ctx context.Context, bucket, object string, rs *HTTPRang
objInfo.Size = rngInfo.Size
rs = nil
}
var nsUnlocker = func() {}
// For a directory, we need to send an reader that returns no bytes.
if HasSuffix(object, SlashSeparator) {
// The lock taken above is released when
// objReader.Close() is called by the caller.
gr, gerr := NewGetObjectReaderFromReader(bytes.NewBuffer(nil), objInfo, opts, nsUnlocker)
gr, gerr := NewGetObjectReaderFromReader(bytes.NewBuffer(nil), objInfo, opts)
return gr, numHits, gerr
}
fn, off, length, nErr := NewGetObjectReader(rs, objInfo, opts, nsUnlocker)
fn, off, length, nErr := NewGetObjectReader(rs, objInfo, opts)
if nErr != nil {
return nil, numHits, nErr
}

21
cmd/erasure-object.go
View file

@ -186,25 +186,30 @@ func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object stri
if objInfo.TransitionStatus == lifecycle.TransitionComplete {
// If transitioned, stream from transition tier unless object is restored locally or restore date is past.
if onDisk := isRestoredObjectOnDisk(objInfo.UserDefined); !onDisk {
return getTransitionedObjectReader(ctx, bucket, object, rs, h, objInfo, opts)
gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, h, objInfo, opts)
if err != nil {
return nil, err
}
unlockOnDefer = false
return gr.WithCleanupFuncs(nsUnlocker), nil
}
}
unlockOnDefer = false
fn, off, length, nErr := NewGetObjectReader(rs, objInfo, opts, nsUnlocker)
if nErr != nil {
return nil, nErr
fn, off, length, err := NewGetObjectReader(rs, objInfo, opts)
if err != nil {
return nil, err
}
unlockOnDefer = false
pr, pw := io.Pipe()
go func() {
err := er.getObjectWithFileInfo(ctx, bucket, object, off, length, pw, fi, metaArr, onlineDisks)
pw.CloseWithError(err)
pw.CloseWithError(er.getObjectWithFileInfo(ctx, bucket, object, off, length, pw, fi, metaArr, onlineDisks))
}()
// Cleanup function to cause the go routine above to exit, in
// case of incomplete read.
pipeCloser := func() { pr.Close() }
return fn(pr, h, opts.CheckPrecondFn, pipeCloser)
return fn(pr, h, opts.CheckPrecondFn, pipeCloser, nsUnlocker)
}
// GetObject - reads an object erasured coded across multiple

6
cmd/fs-v1.go
View file

@ -748,8 +748,10 @@ func (fs *FSObjects) GetObjectNInfo(ctx context.Context, bucket, object string,
rwPoolUnlocker = func() { fs.rwPool.Close(fsMetaPath) }
}
objReaderFn, off, length, err := NewGetObjectReader(rs, objInfo, opts, nsUnlocker, rwPoolUnlocker)
objReaderFn, off, length, err := NewGetObjectReader(rs, objInfo, opts)
if err != nil {
rwPoolUnlocker()
nsUnlocker()
return nil, err
}
@ -777,7 +779,7 @@ func (fs *FSObjects) GetObjectNInfo(ctx context.Context, bucket, object string,
return nil, err
}
return objReaderFn(reader, h, opts.CheckPrecondFn, closeFn)
return objReaderFn(reader, h, opts.CheckPrecondFn, closeFn, rwPoolUnlocker, nsUnlocker)
}
// getObject - wrapper for GetObject

59
cmd/object-api-utils.go
View file

@ -554,14 +554,20 @@ func getCompressedOffsets(objectInfo ObjectInfo, offset int64) (compressedOffset
// GetObjectReader is a type that wraps a reader with a lock to
// provide a ReadCloser interface that unlocks on Close()
type GetObjectReader struct {
ObjInfo ObjectInfo
pReader io.Reader
io.Reader
ObjInfo ObjectInfo
cleanUpFns []func()
opts ObjectOptions
once sync.Once
}
// WithCleanupFuncs sets additional cleanup functions to be called when closing
// the GetObjectReader.
func (g *GetObjectReader) WithCleanupFuncs(fns ...func()) *GetObjectReader {
g.cleanUpFns = append(g.cleanUpFns, fns...)
return g
}
// NewGetObjectReaderFromReader sets up a GetObjectReader with a given
// reader. This ignores any object properties.
func NewGetObjectReaderFromReader(r io.Reader, oi ObjectInfo, opts ObjectOptions, cleanupFns ...func()) (*GetObjectReader, error) {
@ -574,7 +580,7 @@ func NewGetObjectReaderFromReader(r io.Reader, oi ObjectInfo, opts ObjectOptions
}
return &GetObjectReader{
ObjInfo: oi,
pReader: r,
Reader: r,
cleanUpFns: cleanupFns,
opts: opts,
}, nil
@ -587,26 +593,16 @@ func NewGetObjectReaderFromReader(r io.Reader, oi ObjectInfo, opts ObjectOptions
type ObjReaderFn func(inputReader io.Reader, h http.Header, pcfn CheckPreconditionFn, cleanupFns ...func()) (r *GetObjectReader, err error)
// NewGetObjectReader creates a new GetObjectReader. The cleanUpFns
// are called on Close() in reverse order as passed here. NOTE: It is
// are called on Close() in FIFO order as passed in ObjReadFn(). NOTE: It is
// assumed that clean up functions do not panic (otherwise, they may
// not all run!).
func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cleanUpFns ...func()) (
func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions) (
fn ObjReaderFn, off, length int64, err error) {
if rs == nil && opts.PartNumber > 0 {
rs = partNumberToRangeSpec(oi, opts.PartNumber)
}
// Call the clean-up functions immediately in case of exit
// with error
defer func() {
if err != nil {
for i := len(cleanUpFns) - 1; i >= 0; i-- {
cleanUpFns[i]()
}
}
}()
_, isEncrypted := crypto.IsEncrypted(oi.UserDefined)
isCompressed, err := oi.IsCompressedOK()
if err != nil {
@ -658,11 +654,9 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
}
}
fn = func(inputReader io.Reader, h http.Header, pcfn CheckPreconditionFn, cFns ...func()) (r *GetObjectReader, err error) {
cFns = append(cleanUpFns, cFns...)
if opts.CheckPrecondFn != nil && opts.CheckPrecondFn(oi) {
// Call the cleanup funcs
for i := len(cFns) - 1; i >= 0; i-- {
cFns[i]()
for _, cFn := range cFns {
cFn()
}
return nil, PreConditionFailed{}
}
@ -672,8 +666,8 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
inputReader, err = DecryptBlocksRequestR(inputReader, h, 0, firstPart, oi, copySource)
if err != nil {
// Call the cleanup funcs
for i := len(cFns) - 1; i >= 0; i-- {
cFns[i]()
for _, cFn := range cFns {
cFn()
}
return nil, err
}
@ -685,8 +679,8 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
if decOff > 0 {
if err = s2Reader.Skip(decOff); err != nil {
// Call the cleanup funcs
for i := len(cFns) - 1; i >= 0; i-- {
cFns[i]()
for _, cFn := range cFns {
cFn()
}
return nil, err
}
@ -697,9 +691,9 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
rah, err := readahead.NewReaderSize(decReader, compReadAheadBuffers, compReadAheadBufSize)
if err == nil {
decReader = rah
cFns = append(cFns, func() {
cFns = append([]func(){func() {
rah.Close()
})
}}, cFns...)
}
}
oi.Size = decLength
@ -707,7 +701,7 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
// Assemble the GetObjectReader
r = &GetObjectReader{
ObjInfo: oi,
pReader: decReader,
Reader: decReader,
cleanUpFns: cFns,
opts: opts,
}
@ -741,7 +735,6 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
fn = func(inputReader io.Reader, h http.Header, pcfn CheckPreconditionFn, cFns ...func()) (r *GetObjectReader, err error) {
copySource := h.Get(xhttp.AmzServerSideEncryptionCopyCustomerAlgorithm) != ""
cFns = append(cleanUpFns, cFns...)
// Attach decrypter on inputReader
var decReader io.Reader
decReader, err = DecryptBlocksRequestR(inputReader, h, seqNumber, partStart, oi, copySource)
@ -770,7 +763,7 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
// Assemble the GetObjectReader
r = &GetObjectReader{
ObjInfo: oi,
pReader: decReader,
Reader: decReader,
cleanUpFns: cFns,
opts: opts,
}
@ -783,7 +776,6 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
return nil, 0, 0, err
}
fn = func(inputReader io.Reader, _ http.Header, pcfn CheckPreconditionFn, cFns ...func()) (r *GetObjectReader, err error) {
cFns = append(cleanUpFns, cFns...)
if opts.CheckPrecondFn != nil && opts.CheckPrecondFn(oi) {
// Call the cleanup funcs
for i := len(cFns) - 1; i >= 0; i-- {
@ -793,7 +785,7 @@ func NewGetObjectReader(rs *HTTPRangeSpec, oi ObjectInfo, opts ObjectOptions, cl
}
r = &GetObjectReader{
ObjInfo: oi,
pReader: inputReader,
Reader: inputReader,
cleanUpFns: cFns,
opts: opts,
}
@ -815,11 +807,6 @@ func (g *GetObjectReader) Close() error {
return nil
}
// Read - to implement Reader interface.
func (g *GetObjectReader) Read(p []byte) (n int, err error) {
return g.pReader.Read(p)
}
//SealMD5CurrFn seals md5sum with object encryption key and returns sealed
// md5sum
type SealMD5CurrFn func([]byte) []byte