disk_location_ec.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203
  1. package storage
  2. import (
  3. "fmt"
  4. "io/ioutil"
  5. "os"
  6. "path"
  7. "regexp"
  8. "sort"
  9. "strconv"
  10. "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
  11. "github.com/chrislusf/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) FindEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolumeShard, bool) {
  35. l.ecVolumesLock.RLock()
  36. defer l.ecVolumesLock.RUnlock()
  37. ecVolume, ok := l.ecVolumes[vid]
  38. if !ok {
  39. return nil, false
  40. }
  41. for _, ecShard := range ecVolume.Shards {
  42. if ecShard.ShardId == shardId {
  43. return ecShard, true
  44. }
  45. }
  46. return nil, false
  47. }
  48. func (l *DiskLocation) LoadEcShard(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId) (err error) {
  49. ecVolumeShard, err := erasure_coding.NewEcVolumeShard(l.Directory, collection, vid, shardId)
  50. if err != nil {
  51. if err == os.ErrNotExist {
  52. return os.ErrNotExist
  53. }
  54. return fmt.Errorf("failed to create ec shard %d.%d: %v", vid, shardId, err)
  55. }
  56. l.ecVolumesLock.Lock()
  57. defer l.ecVolumesLock.Unlock()
  58. ecVolume, found := l.ecVolumes[vid]
  59. if !found {
  60. ecVolume, err = erasure_coding.NewEcVolume(l.Directory, l.IdxDirectory, collection, vid)
  61. if err != nil {
  62. return fmt.Errorf("failed to create ec volume %d: %v", vid, err)
  63. }
  64. l.ecVolumes[vid] = ecVolume
  65. }
  66. ecVolume.AddEcVolumeShard(ecVolumeShard)
  67. return nil
  68. }
  69. func (l *DiskLocation) UnloadEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) bool {
  70. l.ecVolumesLock.Lock()
  71. defer l.ecVolumesLock.Unlock()
  72. ecVolume, found := l.ecVolumes[vid]
  73. if !found {
  74. return false
  75. }
  76. if _, deleted := ecVolume.DeleteEcVolumeShard(shardId); deleted {
  77. if len(ecVolume.Shards) == 0 {
  78. delete(l.ecVolumes, vid)
  79. ecVolume.Close()
  80. }
  81. return true
  82. }
  83. return true
  84. }
  85. func (l *DiskLocation) loadEcShards(shards []string, collection string, vid needle.VolumeId) (err error) {
  86. for _, shard := range shards {
  87. shardId, err := strconv.ParseInt(path.Ext(shard)[3:], 10, 64)
  88. if err != nil {
  89. return fmt.Errorf("failed to parse ec shard name %v: %v", shard, err)
  90. }
  91. err = l.LoadEcShard(collection, vid, erasure_coding.ShardId(shardId))
  92. if err != nil {
  93. return fmt.Errorf("failed to load ec shard %v: %v", shard, err)
  94. }
  95. }
  96. return nil
  97. }
  98. func (l *DiskLocation) loadAllEcShards() (err error) {
  99. fileInfos, err := ioutil.ReadDir(l.Directory)
  100. if err != nil {
  101. return fmt.Errorf("load all ec shards in dir %s: %v", l.Directory, err)
  102. }
  103. if l.IdxDirectory != l.Directory {
  104. indexFileInfos, err := ioutil.ReadDir(l.IdxDirectory)
  105. if err != nil {
  106. return fmt.Errorf("load all ec shards in dir %s: %v", l.IdxDirectory, err)
  107. }
  108. fileInfos = append(fileInfos, indexFileInfos...)
  109. }
  110. sort.Slice(fileInfos, func(i, j int) bool {
  111. return fileInfos[i].Name() < fileInfos[j].Name()
  112. })
  113. var sameVolumeShards []string
  114. var prevVolumeId needle.VolumeId
  115. for _, fileInfo := range fileInfos {
  116. if fileInfo.IsDir() {
  117. continue
  118. }
  119. ext := path.Ext(fileInfo.Name())
  120. name := fileInfo.Name()
  121. baseName := name[:len(name)-len(ext)]
  122. collection, volumeId, err := parseCollectionVolumeId(baseName)
  123. if err != nil {
  124. continue
  125. }
  126. if re.MatchString(ext) {
  127. if prevVolumeId == 0 || volumeId == prevVolumeId {
  128. sameVolumeShards = append(sameVolumeShards, fileInfo.Name())
  129. } else {
  130. sameVolumeShards = []string{fileInfo.Name()}
  131. }
  132. prevVolumeId = volumeId
  133. continue
  134. }
  135. if ext == ".ecx" && volumeId == prevVolumeId {
  136. if err = l.loadEcShards(sameVolumeShards, collection, volumeId); err != nil {
  137. return fmt.Errorf("loadEcShards collection:%v volumeId:%d : %v", collection, volumeId, err)
  138. }
  139. prevVolumeId = volumeId
  140. continue
  141. }
  142. }
  143. return nil
  144. }
  145. func (l *DiskLocation) deleteEcVolumeById(vid needle.VolumeId) (e error) {
  146. ecVolume, ok := l.ecVolumes[vid]
  147. if !ok {
  148. return
  149. }
  150. ecVolume.Destroy()
  151. delete(l.ecVolumes, vid)
  152. return
  153. }
  154. func (l *DiskLocation) unmountEcVolumeByCollection(collectionName string) map[needle.VolumeId]*erasure_coding.EcVolume {
  155. deltaVols := make(map[needle.VolumeId]*erasure_coding.EcVolume, 0)
  156. for k, v := range l.ecVolumes {
  157. if v.Collection == collectionName {
  158. deltaVols[k] = v
  159. }
  160. }
  161. for k, _ := range deltaVols {
  162. delete(l.ecVolumes, k)
  163. }
  164. return deltaVols
  165. }
  166. func (l *DiskLocation) EcVolumesLen() int {
  167. l.ecVolumesLock.RLock()
  168. defer l.ecVolumesLock.RUnlock()
  169. return len(l.ecVolumes)
  170. }