diff --git a/weed-fs/src/cmd/gstore.go b/weed-fs/src/cmd/gstore.go index 183cbb65c..1567998c4 100644 --- a/weed-fs/src/cmd/gstore.go +++ b/weed-fs/src/cmd/gstore.go @@ -9,10 +9,10 @@ import ( "json" "log" "mime" - "os" "rand" "strconv" "strings" + "time" ) var ( @@ -20,12 +20,12 @@ var ( chunkFolder = flag.String("dir", "/tmp", "data directory to store files") chunkCount = flag.Int("chunks", 5, "data chunks to store files") chunkEnabled = flag.Bool("data", false, "act as a store server") - chunkServer = flag.String("cserver", "localhost:8080", "chunk server to store data") - publicServer = flag.String("pserver", "localhost:8080", "public server to serve data read") - metaServer = flag.String("mserver", "localhost:8080", "metadata server to store mappings") + chunkServer = flag.String("cserver", "localhost:8080", "chunk server to store data") + publicServer = flag.String("pserver", "localhost:8080", "public server to serve data read") + metaServer = flag.String("mserver", "localhost:8080", "metadata server to store mappings") - metaEnabled = flag.Bool("meta", false, "act as a directory server") - metaFolder = flag.String("mdir", "/tmp", "data directory to store mappings") + metaEnabled = flag.Bool("meta", false, "act as a directory server") + metaFolder = flag.String("mdir", "/tmp", "data directory to store mappings") ) type Haystack struct { @@ -57,7 +57,7 @@ func (s *Haystack) GetHandler(w http.ResponseWriter, r *http.Request) { w.Write(n.Data) } func (s *Haystack) PostHandler(w http.ResponseWriter, r *http.Request) { - volumeId, _ := strconv.Atoui64(r.FormValue("volumeId")) + volumeId, _ := strconv.Atoui64(r.FormValue("volumeId")) s.store.Write(volumeId, store.NewNeedle(r)) w.Header().Set("Content-Type", "text/plain") fmt.Fprint(w, "volumeId=", volumeId, "\n") @@ -66,68 +66,74 @@ func (s *Haystack) DeleteHandler(w http.ResponseWriter, r *http.Request) { } func dirReadHandler(w http.ResponseWriter, r *http.Request) { - volumeId, _ := strconv.Atoui64(r.FormValue("volumeId")) - machineList := server.directory.Get((uint32)(volumeId)) - x := rand.Intn(len(machineList)) - machine := machineList[x] - bytes, _ := json.Marshal(machine) - callback := r.FormValue("callback") - w.Header().Set("Content-Type", "application/javascript") - if callback == "" { - w.Write(bytes) - } else { - w.Write([]uint8(callback)) - w.Write([]uint8("(")) - w.Write(bytes) - w.Write([]uint8(")")) - } + volumeId, _ := strconv.Atoui64(r.FormValue("volumeId")) + machineList := server.directory.Get((uint32)(volumeId)) + x := rand.Intn(len(machineList)) + machine := machineList[x] + bytes, _ := json.Marshal(machine) + callback := r.FormValue("callback") + w.Header().Set("Content-Type", "application/javascript") + if callback == "" { + w.Write(bytes) + } else { + w.Write([]uint8(callback)) + w.Write([]uint8("(")) + w.Write(bytes) + w.Write([]uint8(")")) + } } func dirWriteHandler(w http.ResponseWriter, r *http.Request) { - machineList := server.directory.PickForWrite() - bytes, _ := json.Marshal(machineList) - callback := r.FormValue("callback") - w.Header().Set("Content-Type", "application/javascript") - if callback == "" { - w.Write(bytes) - } else { - w.Write([]uint8(callback)) - w.Write([]uint8("(")) - w.Write(bytes) - w.Write([]uint8(")")) - } + machineList := server.directory.PickForWrite() + bytes, _ := json.Marshal(machineList) + callback := r.FormValue("callback") + w.Header().Set("Content-Type", "application/javascript") + if callback == "" { + w.Write(bytes) + } else { + w.Write([]uint8(callback)) + w.Write([]uint8("(")) + w.Write(bytes) + w.Write([]uint8(")")) + } } func dirJoinHandler(w http.ResponseWriter, r *http.Request) { - s := r.FormValue("server") - publicServer := r.FormValue("publicServer") - volumes := make([]store.VolumeStat,0) - json.Unmarshal([]byte(r.FormValue("volumes")), volumes) - server.directory.Add(directory.NewMachine(s,publicServer),volumes) + s := r.FormValue("server") + publicServer := r.FormValue("publicServer") + volumes := make([]store.VolumeStat, 0) + json.Unmarshal([]byte(r.FormValue("volumes")), volumes) + server.directory.Add(directory.NewMachine(s, publicServer), volumes) } var server *Haystack func main() { flag.Parse() + bothEnabled := false if !*chunkEnabled && !*metaEnabled { - fmt.Fprintf(os.Stdout, "Act as both a store server and a directory server\n") + bothEnabled = true + log.Println("Act as both a store server and a directory server") } server = new(Haystack) - if *chunkEnabled { - fmt.Fprintf(os.Stdout, "Chunk data stored in %s\n", *chunkFolder) + if *chunkEnabled || bothEnabled { + log.Println("Chunk data stored in", *chunkFolder) server.store = store.NewStore(*chunkServer, *publicServer, *chunkFolder) defer server.store.Close() http.HandleFunc("/", storeHandler) } - if *metaEnabled { + if *metaEnabled || bothEnabled { server.directory = directory.NewMapper(*metaFolder, "directory") defer server.directory.Save() - http.HandleFunc("/dir/read", dirReadHandler) - http.HandleFunc("/dir/write", dirWriteHandler) - http.HandleFunc("/dir/join", dirJoinHandler) + http.HandleFunc("/dir/read", dirReadHandler) + http.HandleFunc("/dir/write", dirWriteHandler) + http.HandleFunc("/dir/join", dirJoinHandler) } - - server.store.Join(*metaServer) + go func() { + time.Sleep(3000 * 1000) + server.store.Join(*metaServer) + log.Println("stored joined at", *metaServer) + }() log.Println("Serving at http://127.0.0.1:" + strconv.Itoa(*port)) http.ListenAndServe(":"+strconv.Itoa(*port), nil) + } diff --git a/weed-fs/src/pkg/directory/volume_mapping.go b/weed-fs/src/pkg/directory/volume_mapping.go index aba93dc0e..69a8cf7c2 100644 --- a/weed-fs/src/pkg/directory/volume_mapping.go +++ b/weed-fs/src/pkg/directory/volume_mapping.go @@ -31,10 +31,10 @@ func NewMapper(dirname string, filename string) (m *Mapper) { m.fileName = filename log.Println("Loading virtual to physical:", m.dir, "/", m.fileName) dataFile, e := os.OpenFile(m.dir+string(os.PathSeparator)+m.fileName+".map", os.O_RDONLY, 0644) + m.Virtual2physical = make(map[uint32][]*Machine) if e != nil { - log.Fatalf("Mapping File Read [ERROR] %s\n", e) + log.Println("Mapping File Read [ERROR]", e) } else { - m.Virtual2physical = make(map[uint32][]*Machine) decoder := gob.NewDecoder(dataFile) decoder.Decode(m.Virtual2physical) dataFile.Close() @@ -54,13 +54,14 @@ func (m *Mapper) Add(machine *Machine, volumes []store.VolumeStat) { found := false for _, entry := range existing { if machine == entry { - found = true + found = true break } } if !found { - m.Virtual2physical[uint32(v.Id)] = append(existing, machine) + m.Virtual2physical[uint32(v.Id)] = append(existing, machine) } + log.Println(v.Id, "=>", machine.Server) } } func (m *Mapper) Save() { diff --git a/weed-fs/src/pkg/store/store.go b/weed-fs/src/pkg/store/store.go index e9c8b843f..191dbdb24 100644 --- a/weed-fs/src/pkg/store/store.go +++ b/weed-fs/src/pkg/store/store.go @@ -50,11 +50,11 @@ func (s *Store) Join(mserver string) { stats = append(stats, s) } bytes, _ := json.Marshal(stats) - values := new(url.Values) + values := make(url.Values) values.Add("server", s.Server) values.Add("publicServer", s.PublicServer) values.Add("volumes", string(bytes)) - post("http://"+mserver+"/dir/join", *values) + post("http://"+mserver+"/dir/join", values) } func (s *Store) Close() { for _, v := range s.volumes {