Merge pull request #395 from hxiaodon/master

concurrent loading volume when starting volume server

2 files changed, 65 insertions, 24 deletions

diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go
index fc11a411f..e7604a734 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"
 
@@ -21,33 +22,73 @@ func NewDiskLocation(dir string, maxVolumeCount int) *DiskLocation {
 	return location
 }
 
-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)
-						}
-					}
+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)
 				}
 			}
 		}
 	}
+}
+
+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 {
+				task_queue <- dir
+			}
+		}
+		close(task_queue)
+	}()
+
+	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 {
+				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)
 }
 
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
 }