diff options
| -rw-r--r-- | weed/storage/disk_location.go | 61 | ||||
| -rw-r--r-- | 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 } |