gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

perf(objectnode): improve metadata operation performance

Browse Source 
Signed-off-by: tangdeyi <deyitang94@gmail.com>
This commit is contained in:
tangdeyi 2023-07-11 17:11:49 +08:00 committed by jiaruo bai
parent 564b07ebaf
commit 9184d6fcd3
6 changed files with 43 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
docker/s3tests/test_object_head.py
View File

@ -45,12 +45,6 @@ class ObjectHeadTest(S3TestCase):
result=self.s3.delete_object(Bucket=env.BUCKET, Key=key))
self.assert_delete_object_result(
result=self.s3.delete_object(Bucket=env.BUCKET, Key=KEY_PREFIX))
try:
self.s3.head_object(Bucket=env.BUCKET, Key=key)
self.fail() # Non exception occurred is illegal.
except Exception as e:
# Error code 404 is legal.
self.assert_client_error(e, expect_status_code=404)
def test_head_object_if_match(self):
size = 1024 * 256

6
docker/s3tests/test_signature.py
View File

@ -80,12 +80,6 @@ class SignatureTest(S3TestCase):
self.assertEqual(response.status_code, 204)
response.close()
# Check deletion result
url = s3.generate_presigned_url('head_object', Params={'Bucket': env.BUCKET, 'Key': key})
response = requests.head(url)
self.assertEqual(response.status_code, 404)
response.close()
# Remove key prefix
url = s3.generate_presigned_url('delete_object', Params={'Bucket': env.BUCKET, 'Key': KEY_PREFIX})
response = requests.delete(url)

25
docker/s3tests/test_transfer.py
View File

@ -87,27 +87,13 @@ class TransferTest(S3TestCase):
self.assert_delete_object_result(
result=self.s3.delete_object(Bucket=BUCKET, Key=KEY_PREFIX))
def test_transfer_scene1__50mb(self):
def test_transfer_scene(self):
"""
This test tests transfer (upload and download) a 50MB size file by using multipart feature.
:return: None
"""
self.__test_transfer(size=50 * 1024 * 1024)
def test_transfer_scene2_100mb(self):
"""
This test tests transfer (upload and download) a 100MB size file by using multipart feature.
:return: None
"""
self.__test_transfer(size=100 * 1024 * 1024)
def test_transfer_scene3_200mb(self):
"""
This test tests transfer (upload and download) a 200MB size file by using multipart feature.
:return: None
"""
self.__test_transfer(size=200 * 1024 * 1024)
def upload(self, bucket_name, object_name, body):
self.s3.put_object(Bucket=bucket_name,
Key=object_name,
@ -196,15 +182,6 @@ class TransferTest(S3TestCase):
file_size = 30 * 1024 * 1024 # 30M
self.__simulation_java_parallel_download(object_name="java_parallel_download1",part_size_const=part_size, file_size=file_size)
def test_simulation_10_35M(self):
"""
This test tests simulate java sdk to parallel download with part size 10M and file size 35M.
:return: None
"""
part_size = 10 * 1024 * 1024 # 10M
file_size = 35 * 1024 * 1024 # 35M
self.__simulation_java_parallel_download(object_name="java_parallel_download2",part_size_const=part_size, file_size=file_size)
def test_simulation_part_number(self):
"""
This test tests part number is bigger then part count.

4
objectnode/api_handler.go
View File

@ -18,6 +18,7 @@ import (
"errors"
"net/http"
"strings"
"syscall"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/log"
@ -110,6 +111,9 @@ func (o *ObjectNode) getVol(bucket string) (vol *Volume, err error) {
func (o *ObjectNode) errorResponse(w http.ResponseWriter, r *http.Request, err error, ec *ErrorCode) {
if err != nil || ec != nil {
log.LogErrorf("errorResponse: found error: requestID(%v) err(%v) errCode(%v)", GetRequestID(r), err, ec)
if err == syscall.EDQUOT || err == syscall.ENOSPC {
ec = DiskQuotaExceeded
}
if ec1, ok := err.(*ErrorCode); ok && ec == nil {
ec = ec1
}

94
objectnode/fs_volume.go
View File

@ -677,9 +677,6 @@ func (v *Volume) PutObject(path string, reader io.Reader, opt *PutFileOption) (f
log.LogWarnf("PutObject: unlink temp inode: volume(%v) path(%v) inode(%v)",
v.name, path, invisibleTempDataInode.Inode)
_, _ = v.mw.InodeUnlink_ll(invisibleTempDataInode.Inode)
log.LogWarnf("PutObject: evict temp inode: volume(%v) path(%v) inode(%v)",
v.name, path, invisibleTempDataInode.Inode)
_ = v.mw.Evict(invisibleTempDataInode.Inode)
}
}()
@ -910,12 +907,6 @@ func (v *Volume) DeletePath(path string) (err error) {
objMetaCache.DeleteDentry(v.name, dentry.Key())
objMetaCache.DeleteAttr(v.name, ino)
}
log.LogWarnf("DeletePath: evict: volume(%v) path(%v) inode(%v)", v.name, path, ino)
if err = v.mw.Evict(ino); err != nil {
log.LogWarnf("DeletePath Evict: path(%v) inode(%v)", path, ino)
}
err = nil
return
}
@ -1008,18 +999,12 @@ func (v *Volume) WritePart(path string, multipartId string, partId uint16, reade
log.LogWarnf("WritePart: unlink part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartId, partId, tempInodeInfo.Inode)
_, _ = v.mw.InodeUnlink_ll(tempInodeInfo.Inode)
log.LogWarnf("WritePart: evict part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartId, partId, tempInodeInfo.Inode)
_ = v.mw.Evict(tempInodeInfo.Inode)
}
// Delete the old inode and release the written data.
if exist {
log.LogWarnf("WritePart: unlink old part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartId, partId, oldInode)
_, _ = v.mw.InodeUnlink_ll(oldInode)
log.LogWarnf("WritePart: evict old part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartId, partId, oldInode)
_ = v.mw.Evict(oldInode)
}
}()
@ -1097,24 +1082,19 @@ func (v *Volume) AbortMultipart(path string, multipartID string) (err error) {
v.name, multipartID, path, err)
return
}
// release part data
for _, part := range multipartInfo.Parts {
log.LogWarnf("AbortMultipart: unlink part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartID, part.ID, part.Inode)
if _, err = v.mw.InodeUnlink_ll(part.Inode); err != nil {
log.LogErrorf("AbortMultipart: meta inode unlink fail: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v) err(%v)",
v.name, path, multipartID, part.ID, part.Inode, err)
// release part data asyncly
go func() {
for _, part := range multipartInfo.Parts {
log.LogWarnf("AbortMultipart: unlink part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartID, part.ID, part.Inode)
if _, err = v.mw.InodeUnlink_ll(part.Inode); err != nil {
log.LogErrorf("AbortMultipart: meta inode unlink fail: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v) err(%v)",
v.name, path, multipartID, part.ID, part.Inode, err)
}
log.LogDebugf("AbortMultipart: multipart part data released: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartID, part.ID, part.Inode)
}
log.LogWarnf("AbortMultipart: evict part inode: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartID, part.ID, part.Inode)
if err = v.mw.Evict(part.Inode); err != nil {
log.LogErrorf("AbortMultipart: meta inode evict fail: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v) err(%v)",
v.name, path, multipartID, part.ID, part.Inode, err)
}
log.LogDebugf("AbortMultipart: multipart part data released: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, path, multipartID, part.ID, part.Inode)
}
}()
if err = v.mw.RemoveMultipart_ll(path, multipartID); err != nil {
log.LogErrorf("AbortMultipart: meta abort multipart fail: volume(%v) path(%v) multipartID(%v) err(%v)",
v.name, path, multipartID, err)
@ -1269,24 +1249,28 @@ func (v *Volume) CompleteMultipart(path, multipartID string, multipartInfo *prot
log.LogWarnf("CompleteMultipart: remove multipart fail: volume(%v) multipartID(%v) path(%v) err(%v)",
v.name, multipartID, path, err2)
}
// delete part inodes
for _, part := range parts {
log.LogWarnf("CompleteMultipart: destroy part inode: volume(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, multipartID, part.ID, part.Inode)
if err2 = v.mw.InodeDelete_ll(part.Inode); err2 != nil {
log.LogWarnf("CompleteMultipart: delete part inode fail: volume(%v) multipartID(%v) part(%v) err(%v)",
v.name, multipartID, part, err2)
// handle temp metadata asyncly
go func() {
// delete part inodes
for _, part := range parts {
log.LogWarnf("CompleteMultipart: destroy part inode: volume(%v) multipartID(%v) partID(%v) inode(%v)",
v.name, multipartID, part.ID, part.Inode)
if err2 = v.mw.InodeDelete_ll(part.Inode); err2 != nil {
log.LogWarnf("CompleteMultipart: delete part inode fail: volume(%v) multipartID(%v) part(%v) err(%v)",
v.name, multipartID, part, err2)
}
}
}
// discard part inodes
for discardedInode, partNum := range discardedPartInodes {
log.LogWarnf("CompleteMultipart: discard part: volume(%v) multipartID(%v) partNum(%v) inode(%v)",
v.name, multipartID, partNum, discardedInode)
if _, err2 = v.mw.InodeUnlink_ll(discardedInode); err2 != nil {
log.LogWarnf("CompleteMultipart: unlink inode fail: volume(%v) multipartID(%v) inode(%v) err(%v)",
v.name, multipartID, discardedInode, err2)
// discard part inodes and data
for discardedInode, partNum := range discardedPartInodes {
log.LogWarnf("CompleteMultipart: discard part: volume(%v) multipartID(%v) partNum(%v) inode(%v)",
v.name, multipartID, partNum, discardedInode)
if _, err2 = v.mw.InodeUnlink_ll(discardedInode); err2 != nil {
log.LogWarnf("CompleteMultipart: unlink inode fail: volume(%v) multipartID(%v) inode(%v) err(%v)",
v.name, multipartID, discardedInode, err2)
}
}
}
}()
log.LogDebugf("CompleteMultipart: meta complete multipart: volume(%v) multipartID(%v) path(%v) parentID(%v) inode(%v) etagValue(%v)",
v.name, multipartID, path, parentId, finalInode.Inode, etagValue)
@ -1425,6 +1409,11 @@ func (v *Volume) applyInodeToExistDentry(parentID uint64, name string, inode uin
return
}
if oldInode == 0 {
log.LogWarnf("applyInodeToExistDentry: dentry update the same inode: inode(%v)", inode)
return
}
// unlink and evict old inode
log.LogWarnf("applyInodeToExistDentry: unlink inode: volume(%v) inode(%v)", v.name, oldInode)
if _, err = v.mw.InodeUnlink_ll(oldInode); err != nil {
@ -1432,12 +1421,6 @@ func (v *Volume) applyInodeToExistDentry(parentID uint64, name string, inode uin
v.name, oldInode, err)
}
log.LogWarnf("applyInodeToExistDentry: evict inode: volume(%v) inode(%v)", v.name, oldInode)
if err = v.mw.Evict(oldInode); err != nil {
log.LogWarnf("applyInodeToExistDentry: evict inode fail: volume(%v) inode(%v) err(%v)",
v.name, oldInode, err)
}
err = nil
return
}
@ -2657,9 +2640,6 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
log.LogWarnf("CopyFile: unlink target temp inode: volume(%v) path(%v) inode(%v) ",
v.name, targetPath, tInodeInfo.Inode)
_, _ = v.mw.InodeUnlink_ll(tInodeInfo.Inode)
log.LogWarnf("CopyFile: evict target temp inode: volume(%v) path(%v) inode(%v)",
v.name, targetPath, tInodeInfo.Inode)
_ = v.mw.Evict(tInodeInfo.Inode)
}
}()
if err = v.ec.OpenStream(tInodeInfo.Inode); err != nil {

1
objectnode/result_error.go
View File

@ -130,6 +130,7 @@ var (
SignatureDoesNotMatch = &ErrorCode{"SignatureDoesNotMatch", "The request signature we calculated does not match the signature you provided.", http.StatusForbidden}
InvalidMaxPartNumber = &ErrorCode{"InvalidRequest", "the total part numbers exceed limit.", http.StatusBadRequest}
InvalidMinPartNumber = &ErrorCode{"InvalidRequest", "you must specify at least one part.", http.StatusBadRequest}
DiskQuotaExceeded = &ErrorCode{"DiskQuotaExceeded", "Disk Quota Exceeded.", http.StatusBadRequest}
)
func HttpStatusErrorCode(code int) *ErrorCode {