func runFiler(cmd *Command, args []string) bool {

	if err := util.TestFolderWritable(*f.dir); err != nil {
		glog.Fatalf("Check Meta Folder (-dir) Writable %s : %s", *f.dir, err)
	}

	r := http.NewServeMux()
	_, nfs_err := weedserver.NewFilerServer(r, *f.port, *f.master, *f.dir, *f.collection,
		*f.defaultReplicaPlacement, *f.redirectOnRead, *f.disableDirListing,
		*f.secretKey,
		*f.cassandra_server, *f.cassandra_keyspace,
		*f.redis_server, *f.redis_password, *f.redis_database,
	)
	if nfs_err != nil {
		glog.Fatalf("Filer startup error: %v", nfs_err)
	}
	glog.V(0).Infoln("Start Seaweed Filer", util.VERSION, "at port", strconv.Itoa(*f.port))
	filerListener, e := util.NewListener(
		":"+strconv.Itoa(*f.port),
		time.Duration(10)*time.Second,
	)
	if e != nil {
		glog.Fatalf("Filer listener error: %v", e)
	}
	if e := http.Serve(filerListener, r); e != nil {
		glog.Fatalf("Filer Fail to serve: %v", e)
	}

	return true
}

/*
This is to resolve an one-time issue that caused inconsistency with .dat and .idx files.
In this case, the .dat file contains all data, but some of deletion caused incorrect offset.
The .idx has all correct offsets.

1. fix the .dat file, a new .dat_fixed file will be generated.
	go run fix_dat.go -volumeId=9 -dir=/Users/chrislu/Downloads
2. move the original .dat and .idx files to some backup folder, and rename .dat_fixed to .dat file
    mv 9.dat_fixed 9.dat
3. fix the .idx file with the "weed fix"
	weed fix -volumeId=9 -dir=/Users/chrislu/Downloads
*/
func main() {
	flag.Parse()
	fileName := strconv.Itoa(*fixVolumeId)
	if *fixVolumeCollection != "" {
		fileName = *fixVolumeCollection + "_" + fileName
	}
	indexFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".idx"), os.O_RDONLY, 0644)
	if err != nil {
		glog.Fatalf("Read Volume Index [ERROR] %s\n", err)
	}
	defer indexFile.Close()
	datFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".dat"), os.O_RDONLY, 0644)
	if err != nil {
		glog.Fatalf("Read Volume Data [ERROR] %s\n", err)
	}
	defer datFile.Close()

	newDatFile, err := os.Create(path.Join(*fixVolumePath, fileName+".dat_fixed"))
	if err != nil {
		glog.Fatalf("Write New Volume Data [ERROR] %s\n", err)
	}
	defer newDatFile.Close()

	header := make([]byte, storage.SuperBlockSize)
	datFile.Read(header)
	newDatFile.Write(header)

	iterateEntries(datFile, indexFile, func(n *storage.Needle, offset int64) {
		fmt.Printf("file id=%d name=%s size=%d dataSize=%d\n", n.Id, string(n.Name), n.Size, n.DataSize)
		s, e := n.Append(newDatFile, storage.Version2)
		fmt.Printf("size %d error %v\n", s, e)
	})

}

func runMaster(cmd *Command, args []string) bool {
	if *mMaxCpu < 1 {
		*mMaxCpu = runtime.NumCPU()
	}
	runtime.GOMAXPROCS(*mMaxCpu)
	if *masterCpuProfile != "" {
		f, err := os.Create(*masterCpuProfile)
		if err != nil {
			glog.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
		OnInterrupt(func() {
			pprof.StopCPUProfile()
		})
	}
	if err := util.TestFolderWritable(*metaFolder); err != nil {
		glog.Fatalf("Check Meta Folder (-mdir) Writable %s : %s", *metaFolder, err)
	}
	if *masterWhiteListOption != "" {
		masterWhiteList = strings.Split(*masterWhiteListOption, ",")
	}

	r := mux.NewRouter()
	ms := weed_server.NewMasterServer(r, *mport, *metaFolder,
		*volumeSizeLimitMB, *mpulse, *confFile, *defaultReplicaPlacement, *garbageThreshold,
		masterWhiteList, *masterSecureKey,
	)

	listeningAddress := *masterBindIp + ":" + strconv.Itoa(*mport)

	glog.V(0).Infoln("Start Seaweed Master", util.VERSION, "at", listeningAddress)

	listener, e := util.NewListener(listeningAddress, time.Duration(*mTimeout)*time.Second)
	if e != nil {
		glog.Fatalf("Master startup error: %v", e)
	}

	go func() {
		time.Sleep(100 * time.Millisecond)
		myMasterAddress := *masterIp + ":" + strconv.Itoa(*mport)
		var peers []string
		if *masterPeers != "" {
			peers = strings.Split(*masterPeers, ",")
		}
		raftServer := weed_server.NewRaftServer(r, peers, myMasterAddress, *metaFolder, ms.Topo, *mpulse)
		ms.SetRaftServer(raftServer)
	}()

	if e := http.Serve(listener, r); e != nil {
		glog.Fatalf("Fail to serve: %v", e)
	}
	return true
}

func readFileIds(fileName string, fileIdLineChan chan string) {
	file, err := os.Open(fileName) // For read access.
	if err != nil {
		glog.Fatalf("File to read file %s: %s\n", fileName, err)
	}
	defer file.Close()

	r := bufio.NewReader(file)
	if *b.sequentialRead {
		for {
			if line, err := Readln(r); err == nil {
				fileIdLineChan <- string(line)
			} else {
				break
			}
		}
	} else {
		lines := make([]string, 0, readStats.total)
		for {
			if line, err := Readln(r); err == nil {
				lines = append(lines, string(line))
			} else {
				break
			}
		}
		if len(lines) > 0 {
			for i := 0; i < readStats.total; i++ {
				fileIdLineChan <- lines[rand.Intn(len(lines))]
			}
		}
	}

	close(fileIdLineChan)
}

/*
This is to change replication factor in .dat file header. Need to shut down the volume servers
that has those volumes.

1. fix the .dat file in place
	// just see the replication setting
	go run change_replication.go -volumeId=9 -dir=/Users/chrislu/Downloads
		Current Volume Replication: 000
	// fix the replication setting
	go run change_replication.go -volumeId=9 -dir=/Users/chrislu/Downloads -replication 001
		Current Volume Replication: 000
		Changing to: 001
		Done.

2. copy the fixed .dat and related .idx files to some remote server
3. restart volume servers or start new volume servers.
*/
func main() {
	flag.Parse()
	fileName := strconv.Itoa(*fixVolumeId)
	if *fixVolumeCollection != "" {
		fileName = *fixVolumeCollection + "_" + fileName
	}
	datFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".dat"), os.O_RDWR, 0644)
	if err != nil {
		glog.Fatalf("Open Volume Data File [ERROR]: %v", err)
	}
	defer datFile.Close()

	header := make([]byte, storage.SuperBlockSize)
	if _, e := datFile.Read(header); e != nil {
		glog.Fatalf("cannot read volume %s super block: %v", fileName+".dat", e)
	}
	superBlock, err := storage.ParseSuperBlock(header)

	if err != nil {
		glog.Fatalf("cannot parse existing super block: %v", err)
	}

	fmt.Printf("Current Volume Replication: %s\n", superBlock.ReplicaPlacement)

	if *targetReplica == "" {
		return
	}

	replica, err := storage.NewReplicaPlacementFromString(*targetReplica)

	if err != nil {
		glog.Fatalf("cannot parse target replica %s: %v", *targetReplica, err)
	}

	fmt.Printf("Changing to: %s\n", replica)

	superBlock.ReplicaPlacement = replica

	header = superBlock.Bytes()

	if n, e := datFile.WriteAt(header, 0); n == 0 || e != nil {
		glog.Fatalf("cannot write super block: %v", e)
	}

	fmt.Println("Done.")

}

// Loads config information from a JSON string
func LoadConfigString(s string) *Config {
	result := newConfig()
	err := json.Unmarshal([]byte(s), &result.data)
	if err != nil {
		glog.Fatalf("error parsing config string %s: %s", s, err)
	}
	return result
}

func runCompact(cmd *Command, args []string) bool {

	if *compactVolumeId == -1 {
		return false
	}

	vid := storage.VolumeId(*compactVolumeId)
	v, err := storage.NewVolume(*compactVolumePath, *compactVolumeCollection, vid,
		storage.NeedleMapInMemory, nil)
	if err != nil {
		glog.Fatalf("Load Volume [ERROR] %s\n", err)
	}
	if err = v.Compact(); err != nil {
		glog.Fatalf("Compact Volume [ERROR] %s\n", err)
	}

	return true
}

func NewMasterServer(r *mux.Router, port int, metaFolder string,
	volumeSizeLimitMB uint,
	pulseSeconds int,
	confFile string,
	defaultReplicaPlacement string,
	garbageThreshold string,
	whiteList []string,
	secureKey string,
) *MasterServer {
	ms := &MasterServer{
		port:                    port,
		volumeSizeLimitMB:       volumeSizeLimitMB,
		pulseSeconds:            pulseSeconds,
		defaultReplicaPlacement: defaultReplicaPlacement,
		garbageThreshold:        garbageThreshold,
	}
	ms.bounedLeaderChan = make(chan int, 16)
	seq := sequence.NewMemorySequencer()
	cs := storage.NewCollectionSettings(defaultReplicaPlacement, garbageThreshold)
	var e error
	if ms.Topo, e = topology.NewTopology("topo", confFile, cs,
		seq, uint64(volumeSizeLimitMB)*1024*1024, pulseSeconds); e != nil {
		glog.Fatalf("cannot create topology:%s", e)
	}
	ms.vg = topology.NewDefaultVolumeGrowth()
	glog.V(0).Infoln("Volume Size Limit is", volumeSizeLimitMB, "MB")

	ms.guard = security.NewGuard(whiteList, secureKey)

	r.HandleFunc("/", ms.uiStatusHandler)
	r.HandleFunc("/ui/index.html", ms.uiStatusHandler)
	r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(ms.dirAssignHandler)))
	r.HandleFunc("/dir/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.dirLookupHandler)))
	r.HandleFunc("/dir/join", ms.proxyToLeader(ms.guard.WhiteList(ms.dirJoinHandler)))
	r.HandleFunc("/dir/join2", ms.proxyToLeader(ms.guard.WhiteList(ms.dirJoin2Handler)))
	r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(ms.dirStatusHandler)))
	r.HandleFunc("/col/delete", ms.proxyToLeader(ms.guard.WhiteList(ms.collectionDeleteHandler)))
	r.HandleFunc("/vol/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeLookupHandler)))
	r.HandleFunc("/vol/grow", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeGrowHandler)))
	r.HandleFunc("/vol/status", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeStatusHandler)))
	r.HandleFunc("/vol/vacuum", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeVacuumHandler)))
	r.HandleFunc("/vol/check_replicate", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeCheckReplicateHandler)))
	r.HandleFunc("/submit", ms.guard.WhiteList(ms.submitFromMasterServerHandler))
	r.HandleFunc("/delete", ms.guard.WhiteList(ms.deleteFromMasterServerHandler))
	r.HandleFunc("/{fileId}", ms.proxyToLeader(ms.redirectHandler))
	r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
	r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
	r.HandleFunc("/debug/pprof/", ms.guard.WhiteList(pprof.Index))
	r.HandleFunc("/debug/pprof/trace", ms.guard.WhiteList(pprof.Trace))
	r.HandleFunc("/debug/pprof/profile", ms.guard.WhiteList(pprof.Profile))
	r.HandleFunc("/debug/pprof/cmdline", ms.guard.WhiteList(pprof.Cmdline))
	r.HandleFunc("/debug/pprof/symbol", ms.guard.WhiteList(pprof.Symbol))
	r.HandleFunc("/debug/pprof/{name}", ms.guard.WhiteList(pprof.Index))
	ms.Topo.StartRefreshWritableVolumes()

	return ms
}

// Loads config information from a JSON file
func LoadConfig(filename string) *Config {
	result := newConfig()
	result.filename = filename
	err := result.parse()
	if err != nil {
		glog.Fatalf("error loading config file %s: %s", filename, err)
	}
	return result
}

func runFix(cmd *Command, args []string) bool {

	if *fixVolumeId == -1 {
		return false
	}

	fileName := strconv.Itoa(*fixVolumeId)
	if *fixVolumeCollection != "" {
		fileName = *fixVolumeCollection + "_" + fileName
	}
	indexFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".idx"), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	if err != nil {
		glog.Fatalf("Create Volume Index [ERROR] %s\n", err)
	}
	defer indexFile.Close()

	nm := storage.NewNeedleMap(indexFile)
	defer nm.Close()

	vid := storage.VolumeId(*fixVolumeId)
	err = storage.ScanVolumeFile(*fixVolumePath, *fixVolumeCollection, vid,
		storage.NeedleMapInMemory,
		func(superBlock storage.SuperBlock) error {
			return nil
		}, false, func(n *storage.Needle, offset int64) error {
			glog.V(2).Infof("key %d offset %d size %d disk_size %d gzip %v", n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped())
			if n.Size > 0 {
				pe := nm.Put(n.Id, uint32(offset/storage.NeedlePaddingSize), n.Size)
				glog.V(2).Infof("saved %d with error %v", n.Size, pe)
			} else {
				glog.V(2).Infof("skipping deleted file ...")
				return nm.Delete(n.Id)
			}
			return nil
		})
	if err != nil {
		glog.Fatalf("Export Volume File [ERROR] %s\n", err)
	}

	return true
}

func NewFilerServer(r *http.ServeMux, port int, master string, dir string, collection string,
	replication string, redirectOnRead bool, disableDirListing bool,
	secret string,
	cassandra_server string, cassandra_keyspace string,
	redis_server string, redis_password string, redis_database int,
) (fs *FilerServer, err error) {
	fs = &FilerServer{
		master:             master,
		collection:         collection,
		defaultReplication: replication,
		redirectOnRead:     redirectOnRead,
		disableDirListing:  disableDirListing,
		port:               ":" + strconv.Itoa(port),
	}

	if cassandra_server != "" {
		cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server)
		if err != nil {
			glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err)
		}
		fs.filer = flat_namespace.NewFlatNamespaceFiler(master, cassandra_store)
	} else if redis_server != "" {
		redis_store := redis_store.NewRedisStore(redis_server, redis_password, redis_database)
		fs.filer = flat_namespace.NewFlatNamespaceFiler(master, redis_store)
	} else {
		if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil {
			glog.Fatalf("Can not start filer in dir %s : %v", dir, err)
			return
		}

		r.HandleFunc("/admin/mv", fs.moveHandler)
	}

	r.HandleFunc("/", fs.filerHandler)

	return fs, nil
}

/*
This is to see content in .idx files.

	see_idx -v=4 -volumeId=9 -dir=/Users/chrislu/Downloads
*/
func main() {
	flag.Parse()
	fileName := strconv.Itoa(*fixVolumeId)
	if *fixVolumeCollection != "" {
		fileName = *fixVolumeCollection + "_" + fileName
	}
	indexFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".idx"), os.O_RDONLY, 0644)
	if err != nil {
		glog.Fatalf("Create Volume Index [ERROR] %s\n", err)
	}
	defer indexFile.Close()

	storage.LoadNeedleMap(indexFile)

}

func writeFileIds(fileName string, fileIdLineChan chan string, finishChan chan bool) {
	file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	if err != nil {
		glog.Fatalf("File to create file %s: %s\n", fileName, err)
	}
	defer file.Close()

	for {
		select {
		case <-finishChan:
			wait.Done()
			return
		case line := <-fileIdLineChan:
			file.Write([]byte(line))
			file.Write([]byte("\n"))
		}
	}
}

func runExport(cmd *Command, args []string) bool {

	var err error

	if *newer != "" {
		if newerThan, err = time.ParseInLocation(timeFormat, *newer, localLocation); err != nil {
			fmt.Println("cannot parse 'newer' argument: " + err.Error())
			return false
		}
		newerThanUnix = newerThan.Unix()
	}

	if *export.volumeId == -1 {
		return false
	}

	if *output != "" {
		if *output != "-" && !strings.HasSuffix(*output, ".tar") {
			fmt.Println("the output file", *output, "should be '-' or end with .tar")
			return false
		}

		if fileNameTemplate, err = template.New("name").Parse(*format); err != nil {
			fmt.Println("cannot parse format " + *format + ": " + err.Error())
			return false
		}

		var outputFile *os.File
		if *output == "-" {
			outputFile = os.Stdout
		} else {
			if outputFile, err = os.Create(*output); err != nil {
				glog.Fatalf("cannot open output tar %s: %s", *output, err)
			}
		}
		defer outputFile.Close()
		tarOutputFile = tar.NewWriter(outputFile)
		defer tarOutputFile.Close()
		t := time.Now()
		tarHeader = tar.Header{Mode: 0644,
			ModTime: t, Uid: os.Getuid(), Gid: os.Getgid(),
			Typeflag:   tar.TypeReg,
			AccessTime: t, ChangeTime: t}
	}

	fileName := strconv.Itoa(*export.volumeId)
	if *export.collection != "" {
		fileName = *export.collection + "_" + fileName
	}
	vid := storage.VolumeId(*export.volumeId)
	indexFile, err := os.OpenFile(path.Join(*export.dir, fileName+".idx"), os.O_RDONLY, 0644)
	if err != nil {
		glog.Fatalf("Create Volume Index [ERROR] %s\n", err)
	}
	defer indexFile.Close()

	needleMap, err := storage.LoadNeedleMap(indexFile)
	if err != nil {
		glog.Fatalf("cannot load needle map from %s: %s", indexFile.Name(), err)
	}

	var version storage.Version

	err = storage.ScanVolumeFile(*export.dir, *export.collection, vid,
		storage.NeedleMapInMemory,
		func(superBlock storage.SuperBlock) error {
			version = superBlock.Version()
			return nil
		}, true, func(n *storage.Needle, offset int64) error {
			nv, ok := needleMap.Get(n.Id)
			glog.V(3).Infof("key %d offset %d size %d disk_size %d gzip %v ok %v nv %+v",
				n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped(), ok, nv)
			if ok && nv.Size > 0 && int64(nv.Offset)*8 == offset {
				if newerThanUnix >= 0 && n.HasLastModifiedDate() && n.LastModified < uint64(newerThanUnix) {
					glog.V(3).Infof("Skipping this file, as it's old enough: LastModified %d vs %d",
						n.LastModified, newerThanUnix)
					return nil
				}
				return walker(vid, n, version)
			}
			if !ok {
				glog.V(2).Infof("This seems deleted %d size %d", n.Id, n.Size)
			} else {
				glog.V(2).Infof("Skipping later-updated Id %d size %d", n.Id, n.Size)
			}
			return nil
		})
	if err != nil {
		glog.Fatalf("Export Volume File [ERROR] %s\n", err)
	}
	return true
}

func runServer(cmd *Command, args []string) bool {
	filerOptions.secretKey = serverSecureKey
	if *serverOptions.cpuprofile != "" {
		f, err := os.Create(*serverOptions.cpuprofile)
		if err != nil {
			glog.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}

	if *filerOptions.redirectOnRead {
		*isStartingFiler = true
	}

	*filerOptions.master = *serverIp + ":" + strconv.Itoa(*masterPort)

	if *filerOptions.defaultReplicaPlacement == "" {
		*filerOptions.defaultReplicaPlacement = *masterDefaultReplicaPlacement
	}

	if *volumePublicPort == 0 {
		*volumePublicPort = *volumePort
	}

	if *serverMaxCpu < 1 {
		*serverMaxCpu = runtime.NumCPU()
	}
	runtime.GOMAXPROCS(*serverMaxCpu)

	folders := strings.Split(*volumeDataFolders, ",")
	maxCountStrings := strings.Split(*volumeMaxDataVolumeCounts, ",")
	var maxCounts []int
	for _, maxString := range maxCountStrings {
		if max, e := strconv.Atoi(maxString); e == nil {
			maxCounts = append(maxCounts, max)
		} else {
			glog.Fatalf("The max specified in -max not a valid number %s", maxString)
		}
	}
	if len(folders) != len(maxCounts) {
		glog.Fatalf("%d directories by -dir, but only %d max is set by -max", len(folders), len(maxCounts))
	}
	for _, folder := range folders {
		if err := util.TestFolderWritable(folder); err != nil {
			glog.Fatalf("Check Data Folder(-dir) Writable %s : %s", folder, err)
		}
	}

	if *masterMetaFolder == "" {
		*masterMetaFolder = folders[0]
	}
	if *isStartingFiler {
		if *filerOptions.dir == "" {
			*filerOptions.dir = *masterMetaFolder + "/filer"
			os.MkdirAll(*filerOptions.dir, 0700)
		}
		if err := util.TestFolderWritable(*filerOptions.dir); err != nil {
			glog.Fatalf("Check Mapping Meta Folder (-filer.dir=\"%s\") Writable: %s", *filerOptions.dir, err)
		}
	}
	if err := util.TestFolderWritable(*masterMetaFolder); err != nil {
		glog.Fatalf("Check Meta Folder (-mdir=\"%s\") Writable: %s", *masterMetaFolder, err)
	}

	if *serverWhiteListOption != "" {
		serverWhiteList = strings.Split(*serverWhiteListOption, ",")
	}

	if *isStartingFiler {
		go func() {
			r := http.NewServeMux()
			_, nfs_err := weed_server.NewFilerServer(r, *serverBindIp, *filerOptions.port, *filerOptions.master, *filerOptions.dir, *filerOptions.collection,
				*filerOptions.defaultReplicaPlacement,
				*filerOptions.redirectOnRead, *filerOptions.disableDirListing,
				*filerOptions.secretKey,
				*filerOptions.cassandra_server, *filerOptions.cassandra_keyspace,
				*filerOptions.redis_server, *filerOptions.redis_password, *filerOptions.redis_database,
			)
			if nfs_err != nil {
				glog.Fatalf("Filer startup error: %v", nfs_err)
			}
			glog.V(0).Infoln("Start Seaweed Filer", util.VERSION, "at port", strconv.Itoa(*filerOptions.port))
			filerListener, e := util.NewListener(
				":"+strconv.Itoa(*filerOptions.port),
				time.Duration(10)*time.Second,
			)
			if e != nil {
				glog.Fatalf("Filer listener error: %v", e)
			}
			if e := http.Serve(filerListener, r); e != nil {
				glog.Fatalf("Filer Fail to serve: %v", e)
			}
		}()
	}

	var raftWaitForMaster sync.WaitGroup
	var volumeWait sync.WaitGroup

	raftWaitForMaster.Add(1)
//.........这里部分代码省略.........

func runVolume(cmd *Command, args []string) bool {
	if *v.maxCpu < 1 {
		*v.maxCpu = runtime.NumCPU()
	}
	runtime.GOMAXPROCS(*v.maxCpu)

	//Set multiple folders and each folder's max volume count limit'
	v.folders = strings.Split(*volumeFolders, ",")
	maxCountStrings := strings.Split(*maxVolumeCounts, ",")
	for _, maxString := range maxCountStrings {
		if max, e := strconv.Atoi(maxString); e == nil {
			v.folderMaxLimits = append(v.folderMaxLimits, max)
		} else {
			glog.Fatalf("The max specified in -max not a valid number %s", maxString)
		}
	}
	if len(v.folders) != len(v.folderMaxLimits) {
		glog.Fatalf("%d directories by -dir, but only %d max is set by -max", len(v.folders), len(v.folderMaxLimits))
	}
	for _, folder := range v.folders {
		if err := util.TestFolderWritable(folder); err != nil {
			glog.Fatalf("Check Data Folder(-dir) Writable %s : %s", folder, err)
		}
	}

	//security related white list configuration
	if *volumeWhiteListOption != "" {
		v.whiteList = strings.Split(*volumeWhiteListOption, ",")
	}

	//	if *v.ip == "" {
	//		*v.ip = "127.0.0.1"
	//	}

	if *v.publicPort == 0 {
		*v.publicPort = *v.port
	}
	if *v.publicUrl == "" {
		*v.publicUrl = net.JoinHostPort(*v.ip, strconv.Itoa(*v.publicPort))
	}
	isSeperatedPublicPort := *v.publicPort != *v.port

	volumeMux := http.NewServeMux()
	publicVolumeMux := volumeMux
	if isSeperatedPublicPort {
		publicVolumeMux = http.NewServeMux()
	}

	volumeNeedleMapKind := storage.NeedleMapInMemory
	switch *v.indexType {
	case "leveldb":
		volumeNeedleMapKind = storage.NeedleMapLevelDb
	case "boltdb":
		volumeNeedleMapKind = storage.NeedleMapBoltDb
	}
	volumeServer := weedserver.NewVolumeServer(volumeMux, publicVolumeMux,
		*v.ip, *v.port, *v.publicUrl,
		v.folders, v.folderMaxLimits,
		volumeNeedleMapKind,
		*v.master, *v.pulseSeconds, *v.dataCenter, *v.rack,
		v.whiteList,
		*v.fixJpgOrientation, *v.readRedirect, *v.readRemoteNeedle,
	)

	listeningAddress := net.JoinHostPort(*v.bindIp, strconv.Itoa(*v.port))
	glog.V(0).Infoln("Start Seaweed volume server", util.VERSION, "at", listeningAddress)
	listener, e := util.NewListener(listeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second)
	if e != nil {
		glog.Fatalf("Volume server listener error:%v", e)
	}
	if isSeperatedPublicPort {
		publicListeningAddress := net.JoinHostPort(*v.bindIp, strconv.Itoa(*v.publicPort))
		glog.V(0).Infoln("Start Seaweed volume server", util.VERSION, "public at", publicListeningAddress)
		publicListener, e := util.NewListener(publicListeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second)
		if e != nil {
			glog.Fatalf("Volume server listener error:%v", e)
		}
		go func() {
			if e := http.Serve(publicListener, publicVolumeMux); e != nil {
				glog.Fatalf("Volume server fail to serve public: %v", e)
			}
		}()
	}

	OnInterrupt(func() {
		volumeServer.Shutdown()
	})

	if e := http.Serve(listener, volumeMux); e != nil {
		glog.Fatalf("Volume server fail to serve: %v", e)
	}
	return true
}

func NewFilerServer(r *http.ServeMux, ip string, port int, master string, dir string, collection string,
	replication string, redirectOnRead bool, disableDirListing bool,
	confFile string,
	maxMB int,
	secret string,
	cassandra_server string, cassandra_keyspace string,
	redis_server string, redis_password string, redis_database int,
) (fs *FilerServer, err error) {
	fs = &FilerServer{
		master:             master,
		collection:         collection,
		defaultReplication: replication,
		redirectOnRead:     redirectOnRead,
		disableDirListing:  disableDirListing,
		maxMB:              maxMB,
		port:               ip + ":" + strconv.Itoa(port),
	}

	var setting *filerConf
	if confFile != "" {
		setting, err = parseConfFile(confFile)
		if err != nil {
			return nil, err
		}
	} else {
		setting = new(filerConf)
	}

	if setting.MysqlConf != nil && len(setting.MysqlConf) != 0 {
		mysql_store := mysql_store.NewMysqlStore(setting.MysqlConf, setting.IsSharding, setting.ShardCount)
		fs.filer = flat_namespace.NewFlatNamespaceFiler(master, mysql_store)
	} else if cassandra_server != "" {
		cassandra_store, err := cassandra_store.NewCassandraStore(cassandra_keyspace, cassandra_server)
		if err != nil {
			glog.Fatalf("Can not connect to cassandra server %s with keyspace %s: %v", cassandra_server, cassandra_keyspace, err)
		}
		fs.filer = flat_namespace.NewFlatNamespaceFiler(master, cassandra_store)
	} else if redis_server != "" {
		redis_store := redis_store.NewRedisStore(redis_server, redis_password, redis_database)
		fs.filer = flat_namespace.NewFlatNamespaceFiler(master, redis_store)
	} else {
		if fs.filer, err = embedded_filer.NewFilerEmbedded(master, dir); err != nil {
			glog.Fatalf("Can not start filer in dir %s : %v", dir, err)
			return
		}

		r.HandleFunc("/admin/mv", fs.moveHandler)
		r.HandleFunc("/admin/register", fs.registerHandler)
	}

	r.HandleFunc("/", fs.filerHandler)

	go func() {
		connected := true

		fs.masterNodes = storage.NewMasterNodes(fs.master)
		glog.V(0).Infof("Filer server bootstraps with master %s", fs.getMasterNode())

		//force initialize with all available master nodes
		for {
			_, err := fs.masterNodes.FindMaster()
			if err != nil {
				glog.Infof("filer server failed to get master cluster info:%s", err.Error())
				time.Sleep(3 * time.Second)
			} else {
				break
			}
		}

		for {
			glog.V(4).Infof("Filer server sending to master %s", fs.getMasterNode())
			master, err := fs.detectHealthyMaster(fs.getMasterNode())
			if err == nil {
				if !connected {
					connected = true
					if fs.getMasterNode() != master {
						fs.setMasterNode(master)
					}
					glog.V(0).Infoln("Filer Server Connected with master at", master)
				}
			} else {
				glog.V(1).Infof("Filer Server Failed to talk with master %s: %v", fs.getMasterNode(), err)
				if connected {
					connected = false
				}
			}
			if connected {
				time.Sleep(time.Duration(float32(10*1e3)*(1+rand.Float32())) * time.Millisecond)
			} else {
				time.Sleep(time.Duration(float32(10*1e3)*0.25) * time.Millisecond)
			}
		}
	}()

	return fs, nil
}