mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-24 11:09:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor concurrent loading code logic

Browse Source
This commit is contained in:
霍晓栋 2016-11-13 13:24:52 +08:00
parent 096ffa9744
commit b9f385bd66
1 changed files with 45 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
weed/storage/disk_location.go
View File

@ -22,11 +22,43 @@ func NewDiskLocation(dir string, maxVolumeCount int) *DiskLocation {
return location
}
func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) {
l.Lock()
defer l.Unlock()
func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleMapType, mutex *sync.RWMutex) {
name := dir.Name()
if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
collection := ""
base := name[:len(name)-len(".dat")]
i := strings.LastIndex(base, "_")
if i > 0 {
collection, base = base[0:i], base[i+1:]
}
if vid, err := NewVolumeId(base); err == nil {
mutex.RLock()
_, found := l.volumes[vid]
mutex.RUnlock()
if !found {
if v, e := NewVolume(l.Directory, collection, vid, needleMapKind, nil, nil); e == nil {
mutex.Lock()
l.volumes[vid] = v
mutex.Unlock()
glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s", l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), v.Size(), v.Ttl.String())
} else {
glog.V(0).Infof("new volume %s error %s", name, e)
}
}
}
}
}
task_queue := make(chan os.FileInfo, 100)
func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapType, concurrentFlag bool) {
var concurrency int
if concurrentFlag {
//You could choose a better optimized concurency value after testing at your environment
concurrency = 10
} else {
concurrency = 1
}
task_queue := make(chan os.FileInfo, 10*concurrency)
go func() {
if dirs, err := ioutil.ReadDir(l.Directory); err == nil {
for _, dir := range dirs {
@ -36,7 +68,6 @@ func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) {
close(task_queue)
}()
const concurrency int = 10
var wg sync.WaitGroup
var mutex sync.RWMutex
for workerNum := 0; workerNum < concurrency; workerNum++ {
@ -44,35 +75,20 @@ func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) {
go func() {
defer wg.Done()
for dir := range task_queue {
name := dir.Name()
if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
collection := ""
base := name[:len(name)-len(".dat")]
i := strings.LastIndex(base, "_")
if i > 0 {
collection, base = base[0:i], base[i+1:]
}
if vid, err := NewVolumeId(base); err == nil {
mutex.RLock()
_, found := l.volumes[vid]
mutex.RUnlock()
if !found {
if v, e := NewVolume(l.Directory, collection, vid, needleMapKind, nil, nil); e == nil {
mutex.Lock()
l.volumes[vid] = v
mutex.Unlock()
glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s", l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), v.Size(), v.Ttl.String())
} else {
glog.V(0).Infof("new volume %s error %s", name, e)
}
}
}
}
l.loadExistingVolume(dir, needleMapKind, &mutex)
}
}()
}
wg.Wait()
}
func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) {
l.Lock()
defer l.Unlock()
l.concurrentLoadingVolumes(needleMapKind, true)
glog.V(0).Infoln("Store started on dir:", l.Directory, "with", len(l.volumes), "volumes", "max", l.MaxVolumeCount)
}