disk_location_ec.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226
  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "path"
  6. "regexp"
  7. "strconv"
  8. "strings"
  9. "golang.org/x/exp/slices"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
  11. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  12. )
  13. var (
  14. re = regexp.MustCompile(`\.ec[0-9][0-9]`)
  15. )
  16. func (l *DiskLocation) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolume, bool) {
  17. l.ecVolumesLock.RLock()
  18. defer l.ecVolumesLock.RUnlock()
  19. ecVolume, ok := l.ecVolumes[vid]
  20. if ok {
  21. return ecVolume, true
  22. }
  23. return nil, false
  24. }
  25. func (l *DiskLocation) DestroyEcVolume(vid needle.VolumeId) {
  26. l.ecVolumesLock.Lock()
  27. defer l.ecVolumesLock.Unlock()
  28. ecVolume, found := l.ecVolumes[vid]
  29. if found {
  30. ecVolume.Destroy()
  31. delete(l.ecVolumes, vid)
  32. }
  33. }
  34. func (l *DiskLocation) CollectEcShards(vid needle.VolumeId, shardFileNames []string) (ecVolume *erasure_coding.EcVolume, found bool) {
  35. l.ecVolumesLock.RLock()
  36. defer l.ecVolumesLock.RUnlock()
  37. ecVolume, found = l.ecVolumes[vid]
  38. if !found {
  39. return
  40. }
  41. for _, ecShard := range ecVolume.Shards {
  42. if ecShard.ShardId < erasure_coding.ShardId(len(shardFileNames)) {
  43. shardFileNames[ecShard.ShardId] = erasure_coding.EcShardFileName(ecVolume.Collection, l.Directory, int(ecVolume.VolumeId)) + erasure_coding.ToExt(int(ecShard.ShardId))
  44. }
  45. }
  46. return
  47. }
  48. func (l *DiskLocation) FindEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolumeShard, bool) {
  49. l.ecVolumesLock.RLock()
  50. defer l.ecVolumesLock.RUnlock()
  51. ecVolume, ok := l.ecVolumes[vid]
  52. if !ok {
  53. return nil, false
  54. }
  55. for _, ecShard := range ecVolume.Shards {
  56. if ecShard.ShardId == shardId {
  57. return ecShard, true
  58. }
  59. }
  60. return nil, false
  61. }
  62. func (l *DiskLocation) LoadEcShard(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId) (err error) {
  63. ecVolumeShard, err := erasure_coding.NewEcVolumeShard(l.DiskType, l.Directory, collection, vid, shardId)
  64. if err != nil {
  65. if err == os.ErrNotExist {
  66. return os.ErrNotExist
  67. }
  68. return fmt.Errorf("failed to create ec shard %d.%d: %v", vid, shardId, err)
  69. }
  70. l.ecVolumesLock.Lock()
  71. defer l.ecVolumesLock.Unlock()
  72. ecVolume, found := l.ecVolumes[vid]
  73. if !found {
  74. ecVolume, err = erasure_coding.NewEcVolume(l.DiskType, l.Directory, l.IdxDirectory, collection, vid)
  75. if err != nil {
  76. return fmt.Errorf("failed to create ec volume %d: %v", vid, err)
  77. }
  78. l.ecVolumes[vid] = ecVolume
  79. }
  80. ecVolume.AddEcVolumeShard(ecVolumeShard)
  81. return nil
  82. }
  83. func (l *DiskLocation) UnloadEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) bool {
  84. l.ecVolumesLock.Lock()
  85. defer l.ecVolumesLock.Unlock()
  86. ecVolume, found := l.ecVolumes[vid]
  87. if !found {
  88. return false
  89. }
  90. if _, deleted := ecVolume.DeleteEcVolumeShard(shardId); deleted {
  91. if len(ecVolume.Shards) == 0 {
  92. delete(l.ecVolumes, vid)
  93. ecVolume.Close()
  94. }
  95. return true
  96. }
  97. return true
  98. }
  99. func (l *DiskLocation) loadEcShards(shards []string, collection string, vid needle.VolumeId) (err error) {
  100. for _, shard := range shards {
  101. shardId, err := strconv.ParseInt(path.Ext(shard)[3:], 10, 64)
  102. if err != nil {
  103. return fmt.Errorf("failed to parse ec shard name %v: %v", shard, err)
  104. }
  105. err = l.LoadEcShard(collection, vid, erasure_coding.ShardId(shardId))
  106. if err != nil {
  107. return fmt.Errorf("failed to load ec shard %v: %v", shard, err)
  108. }
  109. }
  110. return nil
  111. }
  112. func (l *DiskLocation) loadAllEcShards() (err error) {
  113. dirEntries, err := os.ReadDir(l.Directory)
  114. if err != nil {
  115. return fmt.Errorf("load all ec shards in dir %s: %v", l.Directory, err)
  116. }
  117. if l.IdxDirectory != l.Directory {
  118. indexDirEntries, err := os.ReadDir(l.IdxDirectory)
  119. if err != nil {
  120. return fmt.Errorf("load all ec shards in dir %s: %v", l.IdxDirectory, err)
  121. }
  122. dirEntries = append(dirEntries, indexDirEntries...)
  123. }
  124. slices.SortFunc(dirEntries, func(a, b os.DirEntry) int {
  125. return strings.Compare(a.Name(), b.Name())
  126. })
  127. var sameVolumeShards []string
  128. var prevVolumeId needle.VolumeId
  129. for _, fileInfo := range dirEntries {
  130. if fileInfo.IsDir() {
  131. continue
  132. }
  133. ext := path.Ext(fileInfo.Name())
  134. name := fileInfo.Name()
  135. baseName := name[:len(name)-len(ext)]
  136. collection, volumeId, err := parseCollectionVolumeId(baseName)
  137. if err != nil {
  138. continue
  139. }
  140. info, err := fileInfo.Info()
  141. if err != nil {
  142. continue
  143. }
  144. // 0 byte files should be only appearing erroneously for ec data files
  145. // so we ignore them
  146. if re.MatchString(ext) && info.Size() > 0 {
  147. if prevVolumeId == 0 || volumeId == prevVolumeId {
  148. sameVolumeShards = append(sameVolumeShards, fileInfo.Name())
  149. } else {
  150. sameVolumeShards = []string{fileInfo.Name()}
  151. }
  152. prevVolumeId = volumeId
  153. continue
  154. }
  155. if ext == ".ecx" && volumeId == prevVolumeId {
  156. if err = l.loadEcShards(sameVolumeShards, collection, volumeId); err != nil {
  157. return fmt.Errorf("loadEcShards collection:%v volumeId:%d : %v", collection, volumeId, err)
  158. }
  159. prevVolumeId = volumeId
  160. continue
  161. }
  162. }
  163. return nil
  164. }
  165. func (l *DiskLocation) deleteEcVolumeById(vid needle.VolumeId) (e error) {
  166. ecVolume, ok := l.ecVolumes[vid]
  167. if !ok {
  168. return
  169. }
  170. ecVolume.Destroy()
  171. delete(l.ecVolumes, vid)
  172. return
  173. }
  174. func (l *DiskLocation) unmountEcVolumeByCollection(collectionName string) map[needle.VolumeId]*erasure_coding.EcVolume {
  175. deltaVols := make(map[needle.VolumeId]*erasure_coding.EcVolume, 0)
  176. for k, v := range l.ecVolumes {
  177. if v.Collection == collectionName {
  178. deltaVols[k] = v
  179. }
  180. }
  181. for k, _ := range deltaVols {
  182. delete(l.ecVolumes, k)
  183. }
  184. return deltaVols
  185. }
  186. func (l *DiskLocation) EcVolumesLen() int {
  187. l.ecVolumesLock.RLock()
  188. defer l.ecVolumesLock.RUnlock()
  189. return len(l.ecVolumes)
  190. }