diff --git a/paddle/go/cmd/master/master.go b/paddle/go/cmd/master/master.go new file mode 100644 index 0000000000000000000000000000000000000000..8346b42a329b8d729d4cded02c8fd75a8e73af7a --- /dev/null +++ b/paddle/go/cmd/master/master.go @@ -0,0 +1,78 @@ +package main + +import ( + "flag" + "net" + "net/http" + "net/rpc" + "os" + "strconv" + "strings" + "time" + + "github.com/PaddlePaddle/Paddle/paddle/go/master" + "github.com/wangkuiyi/recordio" +) + +const ( + taskTimeoutDur = 20 * time.Minute + taskTimeoutMax = 3 +) + +func main() { + port := flag.Int("p", 0, "port of the master server") + dataset := flag.String("d", "", "dataset: comma separated path to RecordIO files") + faultTolerant := flag.Bool("fault-tolerance", false, "enable fault tolerance (requires etcd).") + flag.Parse() + + if *dataset == "" { + panic("no dataset specified.") + } + + if *faultTolerant { + panic("fault tolernat not implemented.") + } + + var chunks []master.Chunk + paths := strings.Split(*dataset, ",") + idx := 0 + for _, path := range paths { + f, err := os.Open(path) + if err != nil { + panic(err) + } + + index, err := recordio.LoadIndex(f) + if err != nil { + panic(err) + } + f.Close() + + count := index.NumChunks() + for i := 0; i < count; i++ { + chunk := master.Chunk{ + Idx: idx, + Path: path, + Index: *index.ChunkIndex(i), + } + chunks = append(chunks, chunk) + } + } + + s := master.NewService(chunks, taskTimeoutDur, taskTimeoutMax) + err := rpc.Register(s) + if err != nil { + panic(err) + } + + rpc.HandleHTTP() + l, err := net.Listen("tcp", ":"+strconv.Itoa(*port)) + if err != nil { + panic(err) + } + + err = http.Serve(l, nil) + if err != nil { + panic(err) + } +} diff --git a/paddle/go/master/service.go b/paddle/go/master/service.go index ae7f9687a5ec1d3a9f2a66d6e488a5fc12ef4588..652d345e01d17c26838be5d3d7177f56b351c769 100644 --- a/paddle/go/master/service.go +++ b/paddle/go/master/service.go @@ -64,14 +64,14 @@ func partition(chunks []Chunk, targetTaskCount int) []taskEntry { } // NewService creates a new service. -func NewService(chunks []Chunk, timeoutDur time.Duration, timeoutMax int) (*Service, error) { +func NewService(chunks []Chunk, timeoutDur time.Duration, timeoutMax int) *Service { s := &Service{} s.timeoutDur = timeoutDur s.timeoutMax = timeoutMax s.taskQueues = taskQueues{} s.taskQueues.Pending = make(map[int]taskEntry) s.taskQueues.Todo = partition(chunks, targetTaskCount) - return s, nil + return s } // Chunk is a chunk of data consisted of several data instances.