From 096ffa9744bbca5d9bbed413b6c54ed5da05df39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9C=8D=E6=99=93=E6=A0=8B?= Date: Fri, 11 Nov 2016 11:53:22 +0800 Subject: [PATCH 1/2] concurrent loading volume --- weed/storage/disk_location.go | 61 ++++++++++++++++++++++--------- weed/storage/needle_map_memory.go | 2 +- 2 files changed, 44 insertions(+), 19 deletions(-) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index fc11a411f..d91825c24 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -2,6 +2,7 @@ package storage import ( "io/ioutil" + "os" "strings" "sync" @@ -25,29 +26,53 @@ func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) { l.Lock() defer l.Unlock() - if dirs, err := ioutil.ReadDir(l.Directory); err == nil { - for _, dir := range dirs { - 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 { - if l.volumes[vid] == nil { - if v, e := NewVolume(l.Directory, collection, vid, needleMapKind, nil, nil); e == nil { - l.volumes[vid] = v - 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) + go func() { + if dirs, err := ioutil.ReadDir(l.Directory); err == nil { + for _, dir := range dirs { + task_queue <- dir + } + } + close(task_queue) + }() + + const concurrency int = 10 + var wg sync.WaitGroup + var mutex sync.RWMutex + for workerNum := 0; workerNum < concurrency; workerNum++ { + wg.Add(1) + 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) + } } } } } - } + }() } + wg.Wait() + glog.V(0).Infoln("Store started on dir:", l.Directory, "with", len(l.volumes), "volumes", "max", l.MaxVolumeCount) } diff --git a/weed/storage/needle_map_memory.go b/weed/storage/needle_map_memory.go index f2f4835df..195d8bdbc 100644 --- a/weed/storage/needle_map_memory.go +++ b/weed/storage/needle_map_memory.go @@ -48,7 +48,7 @@ func LoadNeedleMap(file *os.File) (*NeedleMap, error) { } return nil }) - glog.V(1).Infoln("max file key:", nm.MaximumFileKey) + glog.V(1).Infof("max file key: %d for file: %s", nm.MaximumFileKey, file.Name()) return nm, e } From b9f385bd666d256d8b3be130b10e1069c9f231b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9C=8D=E6=99=93=E6=A0=8B?= Date: Sun, 13 Nov 2016 13:24:52 +0800 Subject: [PATCH 2/2] refactor concurrent loading code logic --- weed/storage/disk_location.go | 74 +++++++++++++++++++++-------------- 1 file changed, 45 insertions(+), 29 deletions(-) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index d91825c24..e7604a734 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -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) }