lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-09-23 07:33:34 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
chrislu
2024-07-17 23:13:54 -07:00
parent 7a75928e7d
commit a8fa78b892
2 changed files with 1 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
weed/filer/filer_delete_entry.go
View File

@@ -50,7 +50,7 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
} }
if shouldDeleteChunks && !isDeleteCollection { if shouldDeleteChunks && !isDeleteCollection {
f.DirectDeleteChunks(entry.GetChunks()) f.DeleteChunks(p, entry.GetChunks())
} }
if isDeleteCollection { if isDeleteCollection {

44
weed/filer/filer_deletion.go
View File

@@ -70,50 +70,6 @@ func (f *Filer) loopProcessingDeletion() {
} }
} }
func (f *Filer) doDeleteFileIds(fileIds []string) {
lookupFunc := LookupByMasterClientFn(f.MasterClient)
DeletionBatchSize := 100000 // roughly 20 bytes cost per file id.
for len(fileIds) > 0 {
var toDeleteFileIds []string
if len(fileIds) > DeletionBatchSize {
toDeleteFileIds = fileIds[:DeletionBatchSize]
fileIds = fileIds[DeletionBatchSize:]
} else {
toDeleteFileIds = fileIds
fileIds = fileIds[:0]
}
deletionCount := len(toDeleteFileIds)
_, err := operation.DeleteFilesWithLookupVolumeId(f.GrpcDialOption, toDeleteFileIds, lookupFunc)
if err != nil {
if !strings.Contains(err.Error(), storage.ErrorDeleted.Error()) {
glog.V(0).Infof("deleting fileIds len=%d error: %v", deletionCount, err)
}
}
}
}
func (f *Filer) DirectDeleteChunks(chunks []*filer_pb.FileChunk) {
var fileIdsToDelete []string
for _, chunk := range chunks {
if !chunk.IsChunkManifest {
fileIdsToDelete = append(fileIdsToDelete, chunk.GetFileIdString())
continue
}
dataChunks, manifestResolveErr := ResolveOneChunkManifest(f.MasterClient.LookupFileId, chunk)
if manifestResolveErr != nil {
glog.V(0).Infof("failed to resolve manifest %s: %v", chunk.FileId, manifestResolveErr)
}
for _, dChunk := range dataChunks {
fileIdsToDelete = append(fileIdsToDelete, dChunk.GetFileIdString())
}
fileIdsToDelete = append(fileIdsToDelete, chunk.GetFileIdString())
}
f.doDeleteFileIds(fileIdsToDelete)
}
func (f *Filer) DeleteUncommittedChunks(chunks []*filer_pb.FileChunk) { func (f *Filer) DeleteUncommittedChunks(chunks []*filer_pb.FileChunk) {
f.doDeleteChunks(chunks) f.doDeleteChunks(chunks)
} }