seaweedfs/weed/storage/disk_location_ec.go

193 lines
4.4 KiB
Go
Raw Normal View History

package storage
import (
"fmt"
"io/ioutil"
"path"
"regexp"
"sort"
"strconv"
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
)
var (
re = regexp.MustCompile("\\.ec[0-9][0-9]")
)
2019-05-28 04:40:51 +00:00
func (l *DiskLocation) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolume, bool) {
l.ecVolumesLock.RLock()
defer l.ecVolumesLock.RUnlock()
2019-05-28 04:40:51 +00:00
ecVolume, ok := l.ecVolumes[vid]
if ok {
2019-05-28 04:40:51 +00:00
return ecVolume, true
}
return nil, false
}
2019-06-01 08:51:28 +00:00
func (l *DiskLocation) DestroyEcVolume(vid needle.VolumeId) {
l.ecVolumesLock.Lock()
defer l.ecVolumesLock.Unlock()
ecVolume, found := l.ecVolumes[vid]
if found {
ecVolume.Destroy()
2019-06-05 05:04:10 +00:00
delete(l.ecVolumes, vid)
2019-06-01 08:51:28 +00:00
}
}
2019-05-26 06:23:19 +00:00
func (l *DiskLocation) FindEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolumeShard, bool) {
2019-05-28 04:40:51 +00:00
l.ecVolumesLock.RLock()
defer l.ecVolumesLock.RUnlock()
2019-05-28 04:40:51 +00:00
ecVolume, ok := l.ecVolumes[vid]
2019-05-26 06:23:19 +00:00
if !ok {
return nil, false
}
2019-05-28 04:40:51 +00:00
for _, ecShard := range ecVolume.Shards {
2019-05-26 06:23:19 +00:00
if ecShard.ShardId == shardId {
return ecShard, true
}
}
return nil, false
}
func (l *DiskLocation) LoadEcShard(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId) (err error) {
ecVolumeShard, err := erasure_coding.NewEcVolumeShard(l.Directory, collection, vid, shardId)
if err != nil {
return fmt.Errorf("failed to create ec shard %d.%d: %v", vid, shardId, err)
}
2019-05-28 04:40:51 +00:00
l.ecVolumesLock.Lock()
2019-06-03 09:26:31 +00:00
defer l.ecVolumesLock.Unlock()
2019-05-28 05:00:36 +00:00
ecVolume, found := l.ecVolumes[vid]
if !found {
ecVolume, err = erasure_coding.NewEcVolume(l.Directory, collection, vid)
if err != nil {
return fmt.Errorf("failed to create ec volume %d: %v", vid, err)
}
l.ecVolumes[vid] = ecVolume
}
ecVolume.AddEcVolumeShard(ecVolumeShard)
2019-05-26 06:23:19 +00:00
return nil
}
func (l *DiskLocation) UnloadEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) bool {
2019-05-28 04:40:51 +00:00
l.ecVolumesLock.Lock()
defer l.ecVolumesLock.Unlock()
2019-05-26 06:23:19 +00:00
2019-05-28 04:40:51 +00:00
ecVolume, found := l.ecVolumes[vid]
2019-05-26 06:23:19 +00:00
if !found {
return false
}
if _, deleted := ecVolume.DeleteEcVolumeShard(shardId); deleted {
2019-05-28 04:40:51 +00:00
if len(ecVolume.Shards) == 0 {
delete(l.ecVolumes, vid)
2019-06-06 06:20:26 +00:00
ecVolume.Close()
2019-05-26 06:23:19 +00:00
}
return true
}
return true
}
func (l *DiskLocation) loadEcShards(shards []string, collection string, vid needle.VolumeId) (err error) {
for _, shard := range shards {
shardId, err := strconv.ParseInt(path.Ext(shard)[3:], 10, 64)
if err != nil {
return fmt.Errorf("failed to parse ec shard name %v: %v", shard, err)
}
2019-05-26 06:23:19 +00:00
err = l.LoadEcShard(collection, vid, erasure_coding.ShardId(shardId))
if err != nil {
2019-05-26 06:23:19 +00:00
return fmt.Errorf("failed to load ec shard %v: %v", shard, err)
}
}
return nil
}
2019-05-26 06:23:19 +00:00
func (l *DiskLocation) loadAllEcShards() (err error) {
fileInfos, err := ioutil.ReadDir(l.Directory)
if err != nil {
return fmt.Errorf("load all ec shards in dir %s: %v", l.Directory, err)
}
sort.Slice(fileInfos, func(i, j int) bool {
return fileInfos[i].Name() < fileInfos[j].Name()
})
var sameVolumeShards []string
var prevVolumeId needle.VolumeId
2019-05-26 06:23:19 +00:00
for _, fileInfo := range fileInfos {
if fileInfo.IsDir() {
continue
}
ext := path.Ext(fileInfo.Name())
name := fileInfo.Name()
baseName := name[:len(name)-len(ext)]
collection, volumeId, err := parseCollectionVolumeId(baseName)
if err != nil {
continue
}
2019-05-26 06:23:19 +00:00
if re.MatchString(ext) {
if prevVolumeId == 0 || volumeId == prevVolumeId {
sameVolumeShards = append(sameVolumeShards, fileInfo.Name())
2019-05-26 06:23:19 +00:00
} else {
sameVolumeShards = []string{fileInfo.Name()}
}
prevVolumeId = volumeId
continue
}
2019-05-26 06:23:19 +00:00
if ext == ".ecx" && volumeId == prevVolumeId {
if err = l.loadEcShards(sameVolumeShards, collection, volumeId); err != nil {
return fmt.Errorf("loadEcShards collection:%v volumeId:%d : %v", collection, volumeId, err)
}
prevVolumeId = volumeId
continue
}
}
return nil
}
2019-05-30 16:47:54 +00:00
func (l *DiskLocation) deleteEcVolumeById(vid needle.VolumeId) (e error) {
ecVolume, ok := l.ecVolumes[vid]
if !ok {
return
}
ecVolume.Destroy()
delete(l.ecVolumes, vid)
return
}
func (l *DiskLocation) unmountEcVolumeByCollection(collectionName string) map[needle.VolumeId]*erasure_coding.EcVolume {
deltaVols := make(map[needle.VolumeId]*erasure_coding.EcVolume, 0)
for k, v := range l.ecVolumes {
if v.Collection == collectionName {
deltaVols[k] = v
}
}
for k, _ := range deltaVols {
delete(l.ecVolumes, k)
}
return deltaVols
}
func (l *DiskLocation) EcVolumesLen() int {
l.ecVolumesLock.RLock()
defer l.ecVolumesLock.RUnlock()
return len(l.ecVolumes)
}