fix: add more optimizations to putMetacacheObject() (#12916)

- avoid extra lookup for 'xl.meta' since we are
  definitely sure that it doesn't exist.

- use this in newMultipartUpload() as well

- also additionally do not write with O_DSYNC
  to avoid loading the drives, instead create
  'xl.meta' for listing operations without
  O_DSYNC since these are ephemeral objects.

- do the same with newMultipartUpload() since
  it gets synced when the PutObjectPart() is
  attempted, we do not need to tax newMultipartUpload()
  instead.
This commit is contained in:
Harshavardhana
2021-08-10 11:12:22 -07:00
committed by GitHub
parent 39f81d2c5b
commit 40a2fa8e81
6 changed files with 95 additions and 47 deletions

View File

@@ -677,11 +677,7 @@ func (er erasureObjects) putMetacacheObject(ctx context.Context, key string, r *
onlineDisks[i] = nil
continue
}
if len(inlineBuffers) > 0 && inlineBuffers[i] != nil {
partsMetadata[i].Data = inlineBuffers[i].Bytes()
} else {
partsMetadata[i].Data = nil
}
partsMetadata[i].Data = inlineBuffers[i].Bytes()
partsMetadata[i].AddObjectPart(1, "", n, data.ActualSize())
partsMetadata[i].Erasure.AddChecksumInfo(ChecksumInfo{
PartNumber: 1,
@@ -695,9 +691,10 @@ func (er erasureObjects) putMetacacheObject(ctx context.Context, key string, r *
// Fill all the necessary metadata.
// Update `xl.meta` content on each disks.
for index := range partsMetadata {
partsMetadata[index].Metadata = opts.UserDefined
partsMetadata[index].Size = n
partsMetadata[index].Fresh = true
partsMetadata[index].ModTime = modTime
partsMetadata[index].Metadata = opts.UserDefined
}
// Set an additional header when data is inlined.
@@ -714,33 +711,11 @@ func (er erasureObjects) putMetacacheObject(ctx context.Context, key string, r *
}
}
g := errgroup.WithNErrs(len(onlineDisks))
// Rename file on all underlying storage disks.
for index := range onlineDisks {
index := index
g.Go(func() error {
if onlineDisks[index] == nil {
return errDiskNotFound
}
// Pick one FileInfo for a disk at index.
fi := partsMetadata[index]
// Assign index when index is initialized
if fi.Erasure.Index == 0 {
fi.Erasure.Index = index + 1
}
if fi.IsValid() {
return onlineDisks[index].WriteMetadata(ctx, minioMetaBucket, key, fi)
}
return errFileCorrupt
}, index)
if _, err = writeUniqueFileInfo(ctx, onlineDisks, minioMetaBucket, key, partsMetadata, writeQuorum); err != nil {
return ObjectInfo{}, toObjectErr(err, minioMetaBucket, key)
}
// Wait for all renames to finish.
errs := g.Wait()
return fi.ToObjectInfo(minioMetaBucket, key), reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
return fi.ToObjectInfo(minioMetaBucket, key), nil
}
// PutObject - creates an object upon reading from the input stream