You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

154 lines
4.3 KiB

6 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
  1. package filer
  2. import (
  3. "strings"
  4. "time"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/operation"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. "github.com/chrislusf/seaweedfs/weed/wdclient"
  9. )
  10. func LookupByMasterClientFn(masterClient *wdclient.MasterClient) func(vids []string) (map[string]operation.LookupResult, error) {
  11. return func(vids []string) (map[string]operation.LookupResult, error) {
  12. m := make(map[string]operation.LookupResult)
  13. for _, vid := range vids {
  14. locs, _ := masterClient.GetVidLocations(vid)
  15. var locations []operation.Location
  16. for _, loc := range locs {
  17. locations = append(locations, operation.Location{
  18. Url: loc.Url,
  19. PublicUrl: loc.PublicUrl,
  20. })
  21. }
  22. m[vid] = operation.LookupResult{
  23. VolumeId: vid,
  24. Locations: locations,
  25. }
  26. }
  27. return m, nil
  28. }
  29. }
  30. func (f *Filer) loopProcessingDeletion() {
  31. lookupFunc := LookupByMasterClientFn(f.MasterClient)
  32. DeletionBatchSize := 100000 // roughly 20 bytes cost per file id.
  33. var deletionCount int
  34. for {
  35. deletionCount = 0
  36. f.fileIdDeletionQueue.Consume(func(fileIds []string) {
  37. for len(fileIds) > 0 {
  38. var toDeleteFileIds []string
  39. if len(fileIds) > DeletionBatchSize {
  40. toDeleteFileIds = fileIds[:DeletionBatchSize]
  41. fileIds = fileIds[DeletionBatchSize:]
  42. } else {
  43. toDeleteFileIds = fileIds
  44. fileIds = fileIds[:0]
  45. }
  46. deletionCount = len(toDeleteFileIds)
  47. _, err := operation.DeleteFilesWithLookupVolumeId(f.GrpcDialOption, toDeleteFileIds, lookupFunc)
  48. if err != nil {
  49. if !strings.Contains(err.Error(), "already deleted") {
  50. glog.V(0).Infof("deleting fileIds len=%d error: %v", deletionCount, err)
  51. }
  52. } else {
  53. glog.V(1).Infof("deleting fileIds len=%d", deletionCount)
  54. }
  55. }
  56. })
  57. if deletionCount == 0 {
  58. time.Sleep(1123 * time.Millisecond)
  59. }
  60. }
  61. }
  62. func (f *Filer) doDeleteFileIds(fileIds []string) {
  63. lookupFunc := LookupByMasterClientFn(f.MasterClient)
  64. DeletionBatchSize := 100000 // roughly 20 bytes cost per file id.
  65. for len(fileIds) > 0 {
  66. var toDeleteFileIds []string
  67. if len(fileIds) > DeletionBatchSize {
  68. toDeleteFileIds = fileIds[:DeletionBatchSize]
  69. fileIds = fileIds[DeletionBatchSize:]
  70. } else {
  71. toDeleteFileIds = fileIds
  72. fileIds = fileIds[:0]
  73. }
  74. deletionCount := len(toDeleteFileIds)
  75. _, err := operation.DeleteFilesWithLookupVolumeId(f.GrpcDialOption, toDeleteFileIds, lookupFunc)
  76. if err != nil {
  77. if !strings.Contains(err.Error(), "already deleted") {
  78. glog.V(0).Infof("deleting fileIds len=%d error: %v", deletionCount, err)
  79. }
  80. }
  81. }
  82. }
  83. func (f *Filer) DirectDeleteChunks(chunks []*filer_pb.FileChunk) {
  84. var fildIdsToDelete []string
  85. for _, chunk := range chunks {
  86. if !chunk.IsChunkManifest {
  87. fildIdsToDelete = append(fildIdsToDelete, chunk.GetFileIdString())
  88. continue
  89. }
  90. dataChunks, manifestResolveErr := ResolveOneChunkManifest(f.MasterClient.LookupFileId, chunk)
  91. if manifestResolveErr != nil {
  92. glog.V(0).Infof("failed to resolve manifest %s: %v", chunk.FileId, manifestResolveErr)
  93. }
  94. for _, dChunk := range dataChunks {
  95. fildIdsToDelete = append(fildIdsToDelete, dChunk.GetFileIdString())
  96. }
  97. fildIdsToDelete = append(fildIdsToDelete, chunk.GetFileIdString())
  98. }
  99. f.doDeleteFileIds(fildIdsToDelete)
  100. }
  101. func (f *Filer) DeleteChunks(chunks []*filer_pb.FileChunk) {
  102. for _, chunk := range chunks {
  103. if !chunk.IsChunkManifest {
  104. f.fileIdDeletionQueue.EnQueue(chunk.GetFileIdString())
  105. continue
  106. }
  107. dataChunks, manifestResolveErr := ResolveOneChunkManifest(f.MasterClient.LookupFileId, chunk)
  108. if manifestResolveErr != nil {
  109. glog.V(0).Infof("failed to resolve manifest %s: %v", chunk.FileId, manifestResolveErr)
  110. }
  111. for _, dChunk := range dataChunks {
  112. f.fileIdDeletionQueue.EnQueue(dChunk.GetFileIdString())
  113. }
  114. f.fileIdDeletionQueue.EnQueue(chunk.GetFileIdString())
  115. }
  116. }
  117. func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
  118. if oldEntry == nil {
  119. return
  120. }
  121. if newEntry == nil {
  122. f.DeleteChunks(oldEntry.Chunks)
  123. }
  124. var toDelete []*filer_pb.FileChunk
  125. newChunkIds := make(map[string]bool)
  126. for _, newChunk := range newEntry.Chunks {
  127. newChunkIds[newChunk.GetFileIdString()] = true
  128. }
  129. for _, oldChunk := range oldEntry.Chunks {
  130. if _, found := newChunkIds[oldChunk.GetFileIdString()]; !found {
  131. toDelete = append(toDelete, oldChunk)
  132. }
  133. }
  134. f.DeleteChunks(toDelete)
  135. }