git-svn-id: https://weed-fs.googlecode.com/svn/trunk@6 282b0af5-e82d-9cf1-ede4-77906d7719d0

This commit is contained in:
chris.lu@gmail.com 2011-12-11 20:48:01 +00:00
parent 70be76735c
commit 2858701e6c
3 changed files with 61 additions and 54 deletions

View file

@ -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)
}

View file

@ -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() {

View file

@ -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 {