diff --git a/.script/bin/etcd-v2.0.5-darwin b/.script/bin/etcd-darwin similarity index 57% rename from .script/bin/etcd-v2.0.5-darwin rename to .script/bin/etcd-darwin index 9c6ecb8..c771257 100755 Binary files a/.script/bin/etcd-v2.0.5-darwin and b/.script/bin/etcd-darwin differ diff --git a/.script/bin/etcd-v2.0.5-linux b/.script/bin/etcd-linux similarity index 52% rename from .script/bin/etcd-v2.0.5-linux rename to .script/bin/etcd-linux index 92fd9bc..38abc3f 100755 Binary files a/.script/bin/etcd-v2.0.5-linux and b/.script/bin/etcd-linux differ diff --git a/.script/bin/etcd_version b/.script/bin/etcd_version new file mode 100644 index 0000000..c03bb3d --- /dev/null +++ b/.script/bin/etcd_version @@ -0,0 +1 @@ +v2.0.9 diff --git a/.script/dep.sh b/.script/dep.sh new file mode 100755 index 0000000..7d5bd72 --- /dev/null +++ b/.script/dep.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +# For Go plugin update, please see: +# https://github.com/grpc/grpc-common/tree/master/go + +go get -u github.com/coreos/go-etcd/etcd +go get -u github.com/colinmarc/hdfs +go get -u golang.org/x/net/context +go get -u google.golang.org/grpc +go get -u github.com/golang/protobuf/proto +go get -u github.com/Azure/azure-sdk-for-go/storage + + diff --git a/.script/test b/.script/test index ecbfeb0..aa46e77 100755 --- a/.script/test +++ b/.script/test @@ -1,4 +1,7 @@ -#!/bin/bash -e +#!/bin/bash + +# Exit immediately if a command exits with a non-zero status. +set -e ETCDTESTDIR=".tmp/etcd-testdir" @@ -11,7 +14,7 @@ cleanup() { rm -r $ETCDTESTDIR } # cleanup on any kind of exit -trap "cleanup" SIGINT SIGTERM EXIT +trap cleanup INT TERM EXIT is_etcd_up_on_4001() { if curl -fs "http://localhost:4001/v2/machines" 2>/dev/null; then @@ -30,9 +33,9 @@ if [ -d "$ETCDTESTDIR" ]; then fi if [[ "$OSTYPE" == "darwin"* ]]; then - .script/bin/etcd-v2.0.5-darwin --data-dir $ETCDTESTDIR > /dev/null 2>&1 & + .script/bin/etcd-darwin --data-dir $ETCDTESTDIR > /dev/null 2>&1 & else - .script/bin/etcd-v2.0.5-linux --data-dir $ETCDTESTDIR > /dev/null 2>&1 & + .script/bin/etcd-linux --data-dir $ETCDTESTDIR > /dev/null 2>&1 & fi for i in $(seq 10); do @@ -51,8 +54,8 @@ fi # testing go test -v ./controller -go test -v ./example/regression go test -v ./framework go test -v ./integration go test -v ./op go test -v ./filesystem +go test -v ./example/topo diff --git a/.travis.yml b/.travis.yml index a5a0d37..fece6ed 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,11 +4,7 @@ go: - 1.4 install: - - go get -u github.com/coreos/go-etcd/etcd - - go get -u github.com/colinmarc/hdfs - - go get -u golang.org/x/net/context - - go get -u google.golang.org/grpc - - go get -u github.com/golang/protobuf/proto + - .script/dep.sh before_script: - mkdir .tmp diff --git a/README.md b/README.md index 283763e..c97526f 100644 --- a/README.md +++ b/README.md @@ -16,14 +16,13 @@ on a standard set of events (typed neighbors fail/restart, primary/backup switch to task implementation so that it can do application dependent recovery. -TaskGrpah supports an event driven pull model for communication. When one task +TaskGraph supports an event driven pull model for communication. When one task want to talk to some other task, it set a flag via framework, and framework will notify recipient, which can decide whether or when to fetch data via Framework. Framework will handle communication failures via automatic retry, reconnect to new task node, etc. In another word, it provides exactly-once semantic on data request. - An TaskGraph application usually has three layers. And application implementation need to configure TaskGraph in driver layer and also implement Task/TaskBuilder/Topology based on application logic. diff --git a/controller/controller.go b/controller/controller.go index c1ed7d0..482554b 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -67,10 +67,8 @@ func (c *Controller) InitEtcdLayout() error { for i := uint64(0); i < c.numOfTasks; i++ { key := etcdutil.FreeTaskPath(c.name, strconv.FormatUint(i, 10)) etcdutil.MustCreate(c.etcdclient, c.logger, key, "", 0) - for _, linkType := range c.linkTypes { - key = etcdutil.MetaPath(linkType, c.name, i) - etcdutil.MustCreate(c.etcdclient, c.logger, key, "", 0) - } + key = etcdutil.MetaPath(c.name, i) + etcdutil.MustCreate(c.etcdclient, c.logger, key, "", 0) } return nil } diff --git a/controller/controller_test.go b/controller/controller_test.go index 59a5481..144467e 100644 --- a/controller/controller_test.go +++ b/controller/controller_test.go @@ -29,11 +29,7 @@ func TestControllerInitEtcdLayout(t *testing.T) { if _, err := etcdClient.Get(key, false, false); err != nil { t.Errorf("task %d: etcdClient.Get %v failed: %v", i, key, err) } - key = etcdutil.MetaPath("Parents", c.name, taskID) - if _, err := etcdClient.Get(key, false, false); err != nil { - t.Errorf("task %d: etcdClient.Get %v failed: %v", i, key, err) - } - key = etcdutil.MetaPath("Children", c.name, taskID) + key = etcdutil.MetaPath(c.name, taskID) if _, err := etcdClient.Get(key, false, false); err != nil { t.Errorf("task %d: etcdClient.Get %v failed: %v", i, key, err) } diff --git a/example/bwmf/bwmf_task.go b/example/bwmf/bwmf_task.go deleted file mode 100644 index df26d81..0000000 --- a/example/bwmf/bwmf_task.go +++ /dev/null @@ -1,198 +0,0 @@ -package bwmf - -import ( - "encoding/json" - "log" - "os" - - "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph/pkg/common" -) - -/* TODO: -- SetEpoch 0... -- Topology should be pushed to user. FlagMeta, Meta/Data Ready.. -- Test Data. (Simple, real) -- FileSystem (hdfs, s3).. run real data. -*/ - -/* -The block wise matrix factorization task is designed for carry out block wise matrix -factorization for a variety of criteria (loss function) and constraints (nonnegativity -for example). - -The main idea behind the bwmf is following: -We will have K tasks that handle both row task and column task in alternation. Each task -will read two copies of the data: one row shard and one column shard of A. It either hosts -one shard of D and a full copy of T, or one shard of T and a full copy of D, depending on -the epoch of iteration. "A full copy" consists of computation results from itself and -all "children". -Topology: the topology is different from task to task. Each task will consider itself parent -and all others children. -*/ - -// bwmfData is used to carry indexes and values associated with each index. Index here -// can be row or column id, and value can be K wide, one for each topic; -type bwmfData struct { - Index int - Values []float64 -} - -type Shard []*bwmfData - -func newDShard(index uint64) []*bwmfData { - panic("") -} -func newTShard(index uint64) []*bwmfData { - panic("") -} - -func (shard *Shard) randomFillValue() { -} - -type sparseVec struct { - Indexes []int - Values []float64 -} - -// bwmfTasks holds two shards of original matrices (row and column), one shard of D, -// and one shard of T. It works differently for odd and even epoch: -// During odd epoch, 1. it fetch all T from other slaves, and finding better value for -// local shard of D; 2. after it is done, it let every one knows. Vice versa for even epoch. -// Task 0 will monitor the progress and responsible for starting the work of new epoch. -type bwmfTask struct { - framework taskgraph.Framework - epoch uint64 - taskID uint64 - logger *log.Logger - numOfTasks uint64 - - // The original data. - rowShard, columnShard []sparseVec - - dtReady *common.CountdownLatch - childrenReady map[uint64]bool - - dShard, tShard Shard - d, t []*bwmfData -} - -// These two function carry out actual optimization. -func (t *bwmfTask) updateDShard() {} -func (t *bwmfTask) updateTShard() {} - -// Initialization: We need to read row and column shards of A. -func (t *bwmfTask) readShardsFromDisk() {} - -// Read dShard and tShard from last checkpoint if any. -func (t *bwmfTask) readLastCheckpoint() bool { - panic("") -} - -// Task have all the data, compute local optimization of D/T. -func (t *bwmfTask) localCompute() {} - -// This is useful to bring the task up to speed from scratch or if it recovers. -func (t *bwmfTask) Init(taskID uint64, framework taskgraph.Framework) { - t.taskID = taskID - t.framework = framework - t.logger = log.New(os.Stdout, "", log.Ldate|log.Ltime|log.Lshortfile) - // Use some unique identifier to set Index "in the future". - // We can use taskID now. - t.dShard = newDShard(t.taskID) - t.tShard = newTShard(t.taskID) - t.readShardsFromDisk() - ok := t.readLastCheckpoint() - if !ok { - t.dShard.randomFillValue() - t.tShard.randomFillValue() - } - - // At initialization: - // Task 0 will start the iterations. -} - -func (t *bwmfTask) Exit() {} - -func (t *bwmfTask) SetEpoch(ctx taskgraph.Context, epoch uint64) { - t.logger.Printf("slave SetEpoch, task: %d, epoch: %d\n", t.taskID, epoch) - t.epoch = epoch - t.childrenReady = make(map[uint64]bool) - t.dtReady = common.NewCountdownLatch(int(t.numOfTasks)) - - // Afterwards: - // We need to get all D/T from last epoch so that we can carry out local - // update on T/D. - // Even epochs: Fix D, calculate T; - // Odd epochs: Fix T, calculate D; - - if t.epoch%2 == 0 { - for index := uint64(0); index < t.numOfTasks; index++ { - ctx.DataRequest(index, "getD") - } - } else { - for index := uint64(0); index < t.numOfTasks; index++ { - ctx.DataRequest(index, "getT") - } - } - - go func() { - // Wait for all shards (either D or T, depending on the epoch) to be ready. - t.dtReady.Await() - // We can compute local shard result from A and D/T. - t.localCompute() - // Notify task 0 about the result. - ctx.FlagMetaToParent("computed") - }() -} - -func (t *bwmfTask) ParentMetaReady(ctx taskgraph.Context, parentID uint64, meta string) {} - -func (t *bwmfTask) ParentDataReady(ctx taskgraph.Context, parentID uint64, req string, resp []byte) {} - -func (t *bwmfTask) ChildMetaReady(ctx taskgraph.Context, childID uint64, meta string) { - // Task zero should maintain the barrier for iterations. - if meta == "computed" { - t.childrenReady[childID] = true - } - if uint64(len(t.childrenReady)) < t.numOfTasks { - return - } - // if we have all data, start next iteration. - ctx.IncEpoch() -} - -// Other nodes has served with their local shards. -func (t *bwmfTask) ChildDataReady(ctx taskgraph.Context, childID uint64, req string, resp []byte) { - t.dtReady.CountDown() -} - -// get request of D/T shards from others. Serve with local shard. -func (t *bwmfTask) ServeAsParent(fromID uint64, req string, dataReceiver chan<- []byte) { - var b []byte - var err error - - go func() { - if t.epoch%2 == 0 { - b, err = json.Marshal(t.dShard) - if err != nil { - t.logger.Fatalf("Slave can't encode dShard error: %v\n", err) - } - } else { - b, err = json.Marshal(t.tShard) - if err != nil { - t.logger.Fatalf("Slave can't encode tShard error: %v\n", err) - } - } - dataReceiver <- b - }() -} - -func (t *bwmfTask) ServeAsChild(fromID uint64, req string, dataReceiver chan<- []byte) {} - -type BWMFTaskBuilder struct { -} - -func (tb BWMFTaskBuilder) GetTask(taskID uint64) taskgraph.Task { - return &bwmfTask{} -} diff --git a/example/regression/demo/main.go b/example/regression/demo/main.go index 0302726..6494b56 100644 --- a/example/regression/demo/main.go +++ b/example/regression/demo/main.go @@ -38,7 +38,9 @@ func main() { MasterConfig: map[string]string{"writefile": "result.txt"}, } bootstrap.SetTaskBuilder(taskBuilder) - bootstrap.SetTopology(topo.NewTreeTopology(2, ntask)) + + bootstrap.AddLinkage("Parents" : topo.NewTreeTopologyOfParent(2, ntask)) + bootstrap.AddLinkage("Children" : topo.NewTreeTopologyOfChildren(2, ntask)) bootstrap.Start() default: log.Fatal("Please choose a type: (c) controller, (t) task") diff --git a/example/regression/master.go b/example/regression/master.go index 70ff8ab..c231827 100644 --- a/example/regression/master.go +++ b/example/regression/master.go @@ -18,7 +18,7 @@ import ( // carry out optimization yet. But it should be easy to add support when // this full tests out. // Note: in theory, since there should be no parent of this, so we should -// add error checing in the right places. We will skip these test for now. +// add error checking in the right places. We will skip these test for now. type dummyMaster struct { dataChan chan int32 NodeProducer chan bool @@ -31,6 +31,22 @@ type dummyMaster struct { param *pb.Parameter gradient *pb.Gradient fromChildren map[uint64]*pb.Gradient + + epochChange chan *event + getP chan *event + childDataReady chan *event + exitChan chan struct{} +} + +type event struct { + ctx context.Context + epoch uint64 + input *pb.Input + retP chan *pb.Parameter + retG chan *pb.Gradient + gradient *pb.Gradient + fromID uint64 + output proto.Message } // This is useful to bring the task up to speed from scratch or if it recovers. @@ -38,97 +54,119 @@ func (t *dummyMaster) Init(taskID uint64, framework taskgraph.Framework) { t.taskID = taskID t.framework = framework t.logger = log.New(os.Stdout, "", log.Ldate|log.Ltime|log.Lshortfile) + + t.epochChange = make(chan *event, 1) + t.getP = make(chan *event, 1) + t.childDataReady = make(chan *event, 1) + t.exitChan = make(chan struct{}) + go t.run() } -func (t *dummyMaster) Exit() {} - -// Ideally, we should also have the following: -func (t *dummyMaster) MetaReady(ctx context.Context, fromID uint64, linkType, meta string) { - if linkType == "Children" { - t.logger.Printf("master ChildMetaReady, task: %d, epoch: %d, child: %d\n", t.taskID, t.epoch, fromID) - // Get data from child. When all the data is back, starts the next epoch. - switch meta { - case "GradientReady": - t.framework.DataRequest(ctx, fromID, "/proto.Regression/GetGradient", &pb.Input{t.epoch}) - default: - panic("") + +func (t *dummyMaster) run() { + for { + select { + case ec := <-t.epochChange: + t.enterEpoch(ec.ctx, ec.epoch) + case req := <-t.getP: + // We have to check epoch here in user level because grpc doesn't + // allow use to intercept messages. This should be fixed later. + err := t.framework.CheckGRPCContext(req.ctx) + if err != nil { + close(req.retP) + break + } + req.retP <- t.param + case cr := <-t.childDataReady: + t.ChildDataReady(cr.ctx, cr.fromID, cr.output) + case <-t.exitChan: + return } } } +func (t *dummyMaster) Exit() { + close(t.exitChan) +} + // This give the task an opportunity to cleanup and regroup. -func (t *dummyMaster) SetEpoch(ctx context.Context, epoch uint64) { - t.logger.Printf("master SetEpoch, task: %d, epoch: %d\n", t.taskID, epoch) +func (t *dummyMaster) EnterEpoch(ctx context.Context, epoch uint64) { if t.testablyFail("SetEpoch", strconv.FormatUint(epoch, 10)) { return } + t.epochChange <- &event{ctx: ctx, epoch: epoch} +} +func (t *dummyMaster) enterEpoch(ctx context.Context, epoch uint64) { + t.logger.Printf("master EnterEpoch, task %d, epoch %d\n", t.taskID, epoch) t.param = new(pb.Parameter) - t.gradient = new(pb.Gradient) + t.fromChildren = make(map[uint64]*pb.Gradient) t.epoch = epoch t.param.Value = int32(t.epoch) - - // Make sure we have a clean slate. - t.fromChildren = make(map[uint64]*pb.Gradient) - t.framework.FlagMeta(ctx, "Parents", "ParamReady") + for _, c := range t.framework.GetTopology()["Children"].GetNeighbors(epoch) { + t.framework.DataRequest(ctx, c, "/proto.Regression/GetGradient", &pb.Input{}) + } } -// These are payload rpc for application purpose. func (t *dummyMaster) GetParameter(ctx context.Context, input *pb.Input) (*pb.Parameter, error) { - err := t.framework.CheckEpoch(input.Epoch) - if err != nil { - return nil, err + retP := make(chan *pb.Parameter, 1) + t.getP <- &event{ctx: ctx, input: input, retP: retP} + p, ok := <-retP + if !ok { + return nil, fmt.Errorf("epoch changed") } - return t.param, nil + return p, nil } func (t *dummyMaster) GetGradient(ctx context.Context, input *pb.Input) (*pb.Gradient, error) { panic("") } +func (t *dummyMaster) DataReady(ctx context.Context, fromID uint64, method string, output proto.Message) { + if method == "/proto.Regression/GetGradient" { + t.childDataReady <- &event{ctx: ctx, fromID: fromID, output: output} + return + } + panic("") +} + +func (t *dummyMaster) gradientReady(ctx context.Context) { + // In testing, we need to make sure dataChan has enough space and don't block. + t.dataChan <- t.gradient.Value + // In real ML, we modify the gradient first. But here it is noop. + if t.epoch == t.numberOfIterations { + if t.config["writefile"] != "" { + data := []byte(fmt.Sprintf("Finished job. Gradient value %v\n", t.gradient.Value)) + ioutil.WriteFile(t.config["writefile"], data, 0644) + } + t.framework.ShutdownJob() + } else { + t.logger.Printf("master finished current epoch, task %d, epoch %d", t.taskID, t.epoch) + t.framework.IncEpoch(ctx) + } +} func (t *dummyMaster) ChildDataReady(ctx context.Context, childID uint64, output proto.Message) { d, ok := output.(*pb.Gradient) if !ok { - panic("") + t.logger.Fatalf("Can't convert proto message to Gradient: %v", output) } t.fromChildren[childID] = d - t.logger.Printf("master ChildDataReady, task: %d, epoch: %d, child: %d, ready: %d\n", + t.logger.Printf("master ChildDataReady, task %d, epoch %d, child %d, ready %d\n", t.taskID, t.epoch, childID, len(t.fromChildren)) // This is a weak form of checking. We can also check the task ids. // But this really means that we get all the events from children, we // should go into the next epoch now. - if len(t.fromChildren) == len(t.framework.GetTopology().GetNeighbors("Children", t.epoch)) { + if len(t.fromChildren) == len(t.framework.GetTopology()["Children"].GetNeighbors(t.epoch)) { + t.gradient = new(pb.Gradient) for _, g := range t.fromChildren { t.gradient.Value += g.Value } - - t.dataChan <- t.gradient.Value - // TODO(xiaoyunwu) We need to do some test here. - - // In real ML, we modify the gradient first. But here it is noop. - if t.epoch == t.numberOfIterations { - if t.config["writefile"] != "" { - data := []byte(fmt.Sprintf("Finished job. Gradient value: %v\n", t.gradient.Value)) - ioutil.WriteFile(t.config["writefile"], data, 0644) - } - t.framework.ShutdownJob() - } else { - t.logger.Printf("master finished current epoch, task: %d, epoch: %d", t.taskID, t.epoch) - t.framework.IncEpoch(ctx) - } + t.gradientReady(ctx) } } -func (t *dummyMaster) DataReady(ctx context.Context, fromID uint64, method string, output proto.Message) { - if method == "/proto.Regression/GetGradient" { - t.ChildDataReady(ctx, fromID, output) - return - } - panic("") -} - func (t *dummyMaster) CreateOutputMessage(methodName string) proto.Message { switch methodName { case "/proto.Regression/GetParameter": @@ -163,8 +201,11 @@ func (t *dummyMaster) testablyFail(method string, args ...string) bool { if !probablyFail(t.config["faillevel"]) { return false } - t.logger.Printf("master task %d testably fail, method: %s\n", t.taskID, method) + t.logger.Printf("master task %d testably fail, method %s\n", t.taskID, method) + // Very hack. Need some internal knowledge. Don't change this. t.framework.Kill() t.NodeProducer <- true return true } + +func (t *dummyMaster) MetaReady(ctx context.Context, fromID uint64, linkType, meta string) {} diff --git a/example/regression/proto/regression.pb.go b/example/regression/proto/regression.pb.go index 06965fa..e0f9663 100644 --- a/example/regression/proto/regression.pb.go +++ b/example/regression/proto/regression.pb.go @@ -30,7 +30,6 @@ var _ grpc.ClientConn var _ = proto1.Marshal type Input struct { - Epoch uint64 `protobuf:"varint,1,opt,name=epoch" json:"epoch,omitempty"` } func (m *Input) Reset() { *m = Input{} } @@ -100,9 +99,9 @@ func RegisterRegressionServer(s *grpc.Server, srv RegressionServer) { s.RegisterService(&_Regression_serviceDesc, srv) } -func _Regression_GetParameter_Handler(srv interface{}, ctx context.Context, buf []byte) (proto1.Message, error) { +func _Regression_GetParameter_Handler(srv interface{}, ctx context.Context, codec grpc.Codec, buf []byte) (interface{}, error) { in := new(Input) - if err := proto1.Unmarshal(buf, in); err != nil { + if err := codec.Unmarshal(buf, in); err != nil { return nil, err } out, err := srv.(RegressionServer).GetParameter(ctx, in) @@ -112,9 +111,9 @@ func _Regression_GetParameter_Handler(srv interface{}, ctx context.Context, buf return out, nil } -func _Regression_GetGradient_Handler(srv interface{}, ctx context.Context, buf []byte) (proto1.Message, error) { +func _Regression_GetGradient_Handler(srv interface{}, ctx context.Context, codec grpc.Codec, buf []byte) (interface{}, error) { in := new(Input) - if err := proto1.Unmarshal(buf, in); err != nil { + if err := codec.Unmarshal(buf, in); err != nil { return nil, err } out, err := srv.(RegressionServer).GetGradient(ctx, in) diff --git a/example/regression/proto/regression.proto b/example/regression/proto/regression.proto index efafcf5..a646402 100644 --- a/example/regression/proto/regression.proto +++ b/example/regression/proto/regression.proto @@ -8,7 +8,6 @@ service Regression { } message Input { - uint64 epoch = 1; } message Parameter { diff --git a/example/regression/slave.go b/example/regression/slave.go index a5f3684..38820a0 100644 --- a/example/regression/slave.go +++ b/example/regression/slave.go @@ -1,15 +1,16 @@ package regression import ( + "fmt" "log" "os" "github.com/golang/protobuf/proto" "github.com/taskgraph/taskgraph" pb "github.com/taskgraph/taskgraph/example/regression/proto" - "github.com/taskgraph/taskgraph/pkg/common" "golang.org/x/net/context" "google.golang.org/grpc" + "google.golang.org/grpc/metadata" ) // dummySlave is an prototype for data shard in machine learning applications. @@ -22,11 +23,18 @@ type dummySlave struct { NodeProducer chan bool config map[string]string - param *pb.Parameter - gradient *pb.Gradient - fromChildren map[uint64]*pb.Gradient - gradientReady *common.CountdownLatch - parameterReady *common.CountdownLatch + param *pb.Parameter + gradient *pb.Gradient + fromChildren map[uint64]*pb.Gradient + + epochChange chan *event + getP chan *event + getG chan *event + pDataReady chan *event + gDataReady chan *event + getPReqs []*event + getGReqs []*event + exitChan chan struct{} } // This is useful to bring the task up to speed from scratch or if it recovers. @@ -34,136 +42,184 @@ func (t *dummySlave) Init(taskID uint64, framework taskgraph.Framework) { t.taskID = taskID t.framework = framework t.logger = log.New(os.Stdout, "", log.Ldate|log.Ltime|log.Lshortfile) + + t.epochChange = make(chan *event, 1) + t.getP = make(chan *event, 1) + t.getG = make(chan *event, 1) + t.pDataReady = make(chan *event, 1) + t.gDataReady = make(chan *event, 1) + t.exitChan = make(chan struct{}) + go t.run() } -func (t *dummySlave) Exit() {} -// Ideally, we should also have the following -func (t *dummySlave) MetaReady(ctx context.Context, fromID uint64, linkType, meta string) { - if linkType == "Parents" { - t.logger.Printf("slave ParentMetaReady, task: %d, epoch: %d\n", t.taskID, t.epoch) - t.framework.DataRequest(ctx, fromID, "/proto.Regression/GetParameter", &pb.Input{t.epoch}) - } - if linkType == "Children" { - t.logger.Printf("slave ChildMetaReady, task: %d, epoch: %d\n", t.taskID, t.epoch) - go func() { - // If a new node restart and find out both parent and child meta ready, it will - // simultaneously request both data. We need to wait until gradient data is there. - t.gradientReady.Await() - t.framework.DataRequest(ctx, fromID, "/proto.Regression/GetGradient", &pb.Input{t.epoch}) - }() +func (t *dummySlave) run() { + for { + select { + case ec := <-t.epochChange: + for _, req := range t.getPReqs { + close(req.retP) + } + t.getPReqs = nil + for _, req := range t.getGReqs { + close(req.retG) + } + t.getGReqs = nil + + t.enterEpoch(ec.ctx, ec.epoch) + case req := <-t.getP: + // We have to check epoch here in user level because grpc doesn't + // allow use to intercept messages. This should be fixed later. + err := t.framework.CheckGRPCContext(req.ctx) + if err != nil { + close(req.retP) + break + } + if t.param != nil { + req.retP <- t.param + break + } + // Waiting queue. Requests will get notified later. The number of request + // won't be huge presumingly. + t.getPReqs = append(t.getPReqs, req) + case req := <-t.getG: + err := t.framework.CheckGRPCContext(req.ctx) + if err != nil { + close(req.retG) + break + } + if t.gradient != nil { + req.retG <- t.gradient + break + } + // Waiting queue. Requests will get notified later. The number of request + // won't be huge presumingly. + t.getGReqs = append(t.getGReqs, req) + case pr := <-t.pDataReady: + t.ParentDataReady(pr.ctx, pr.fromID, pr.output) + case gr := <-t.gDataReady: + t.ChildDataReady(gr.ctx, gr.fromID, gr.output) + case <-t.exitChan: + return + } } } -// This give the task an opportunity to cleanup and regroup. -func (t *dummySlave) SetEpoch(ctx context.Context, epoch uint64) { - t.logger.Printf("slave SetEpoch, task: %d, epoch: %d\n", t.taskID, epoch) +func (t *dummySlave) Exit() { + close(t.exitChan) +} - t.param = new(pb.Parameter) - t.gradient = new(pb.Gradient) - t.gradientReady = common.NewCountdownLatch(1) - t.parameterReady = common.NewCountdownLatch(1) +// This give the task an opportunity to cleanup and regroup. +func (t *dummySlave) EnterEpoch(ctx context.Context, epoch uint64) { + t.epochChange <- &event{ctx: ctx, epoch: epoch} +} - t.epoch = epoch - // Make sure we have a clean slate. +func (t *dummySlave) enterEpoch(ctx context.Context, epoch uint64) { + t.logger.Printf("slave EnterEpoch, task %d, epoch %d\n", t.taskID, epoch) + t.param = nil + t.gradient = nil t.fromChildren = make(map[uint64]*pb.Gradient) + t.epoch = epoch + + parent := t.framework.GetTopology()["Parents"].GetNeighbors(epoch)[0] + t.framework.DataRequest(ctx, parent, "/proto.Regression/GetParameter", &pb.Input{}) + + for _, c := range t.framework.GetTopology()["Children"].GetNeighbors(t.epoch) { + t.framework.DataRequest(ctx, c, "/proto.Regression/GetGradient", &pb.Input{}) + } } func (t *dummySlave) GetParameter(ctx context.Context, input *pb.Input) (*pb.Parameter, error) { - err := t.framework.CheckEpoch(input.Epoch) - if err != nil { - return nil, err + retP := make(chan *pb.Parameter, 1) + t.getP <- &event{ctx: ctx, input: input, retP: retP} + p, ok := <-retP + if !ok { + return nil, fmt.Errorf("epoch changed") } - // There is a race: - // A -> B -> C (parent -> child) - // B has flagged "parameter Ready" - // Now B crashed, and C crashed. C restarted and found B has flagged "parameter Ready". - // C requested B. B needs to await until it actually has the data. - t.parameterReady.Await() - return t.param, nil + md, _ := metadata.FromContext(ctx) + t.logger.Printf("slave serve GetParameter, task %d, from %s, epoch %s", t.taskID, md["taskID"], md["epoch"]) + return p, nil } func (t *dummySlave) GetGradient(ctx context.Context, input *pb.Input) (*pb.Gradient, error) { - err := t.framework.CheckEpoch(input.Epoch) - if err != nil { - return nil, err + retG := make(chan *pb.Gradient, 1) + t.getG <- &event{ctx: ctx, input: input, retG: retG} + g, ok := <-retG + if !ok { + return nil, fmt.Errorf("epoch changed") } - return t.gradient, nil + md, _ := metadata.FromContext(ctx) + t.logger.Printf("slave serve GetGradient, task %d, from %s, epoch %s", t.taskID, md["taskID"], md["epoch"]) + return g, nil } -func (t *dummySlave) ParentDataReady(ctx context.Context, parentID uint64, output proto.Message) { - t.logger.Printf("slave ParentDataReady, task: %d, epoch: %d, parent: %d\n", t.taskID, t.epoch, parentID) - if t.testablyFail("ParentDataReady") { +func (t *dummySlave) parameterReady() { + for _, req := range t.getPReqs { + req.retP <- t.param + } + t.getPReqs = nil +} +func (t *dummySlave) gradientReady(ctx context.Context) { + t.logger.Printf("slave gradient ready, task %d, epoch %d, gradient %d", t.taskID, t.epoch, t.gradient.Value) + // If this failure happens, a new node will redo computing again. + if t.testablyFail("ChildDataReady") { + return + } + for _, req := range t.getGReqs { + req.retG <- t.gradient + } + t.getGReqs = nil + // if this failure happens, the parent could + // 1. not get the data. DataRequest shall retry. + // 2. already get the data. Where everything will continue and some work will drop + // if epoch bumps up. + if t.testablyFail("ChildDataReady") { return } - if t.gradientReady.Count() == 0 { +} +func (t *dummySlave) checkGradReady(ctx context.Context) { + children := t.framework.GetTopology()["Children"].GetNeighbors(t.epoch) + if t.param != nil && len(t.fromChildren) == len(children) { + t.gradient = new(pb.Gradient) + t.gradient.Value = t.param.Value * int32(t.taskID) + // In real ML, we add the gradient first. + for _, g := range t.fromChildren { + t.gradient.Value += g.Value + } + t.gradientReady(ctx) + } +} + +func (t *dummySlave) ParentDataReady(ctx context.Context, parentID uint64, output proto.Message) { + t.logger.Printf("slave ParentDataReady, task %d, epoch %d, parent %d\n", t.taskID, t.epoch, parentID) + if t.testablyFail("ParentDataReady") { return } d, ok := output.(*pb.Parameter) if !ok { - panic("") + t.logger.Fatalf("Can't convert proto message to Gradient: %v", output) } t.param = d - t.parameterReady.CountDown() - // We need to carry out local compuation. - t.gradient.Value = t.param.Value * int32(t.framework.GetTaskID()) - t.gradientReady.CountDown() - - // If this task has children, flag meta so that children can start pull - // parameter. - children := t.framework.GetTopology().GetNeighbors("Children", t.epoch) - if len(children) != 0 { - t.framework.FlagMeta(ctx, "Parents", "ParamReady") - } else { - // On leaf node, we can immediately return by and flag parent - // that this node is ready. - t.framework.FlagMeta(ctx, "Children", "GradientReady") - } + t.parameterReady() + t.checkGradReady(ctx) } func (t *dummySlave) ChildDataReady(ctx context.Context, childID uint64, output proto.Message) { d, ok := output.(*pb.Gradient) if !ok { - panic("") + t.logger.Fatalf("Can't convert proto message to Gradient: %v", output) } t.fromChildren[childID] = d - t.logger.Printf("slave ChildDataReady, task: %d, epoch: %d, child: %d, ready: %d\n", + t.logger.Printf("slave ChildDataReady, task %d, epoch %d, child %d, ready %d\n", t.taskID, t.epoch, childID, len(t.fromChildren)) - - // This is a weak form of checking. We can also check the task ids. - // But this really means that we get all the events from children, we - // should go into the next epoch now. - if len(t.fromChildren) == len(t.framework.GetTopology().GetNeighbors("Children", t.epoch)) { - // In real ML, we add the gradient first. - for _, g := range t.fromChildren { - t.gradient.Value += g.Value - } - - // If this failure happens, a new node will redo computing again. - if t.testablyFail("ChildDataReady") { - return - } - - t.framework.FlagMeta(ctx, "Children", "GradientReady") - - // if this failure happens, the parent could - // 1. not have the data yet. In such case, the parent could - // 1.1 not request the data before a new node restarts. This will cause - // double requests since we provide at-least-once semantics (!outdated). - // 1.2 request the data with a failed host (request should fail or be - // responded with error message). - // 2. already get the data. - if t.testablyFail("ChildDataReady") { - return - } - } + t.checkGradReady(ctx) } func (t *dummySlave) DataReady(ctx context.Context, fromID uint64, method string, output proto.Message) { if method == "/proto.Regression/GetParameter" { - t.ParentDataReady(ctx, fromID, output) + t.pDataReady <- &event{ctx: ctx, fromID: fromID, output: output} } else { - t.ChildDataReady(ctx, fromID, output) + t.gDataReady <- &event{ctx: ctx, fromID: fromID, output: output} } } @@ -178,6 +234,7 @@ func (t *dummySlave) testablyFail(method string, args ...string) bool { return false } t.logger.Printf("slave task %d testably fail, method: %s\n", t.taskID, method) + // Very hack. Need some internal knowledge. Don't change this. t.framework.Kill() t.NodeProducer <- true return true @@ -199,3 +256,5 @@ func (t *dummySlave) CreateServer() *grpc.Server { pb.RegisterRegressionServer(server, t) return server } + +func (t *dummySlave) MetaReady(ctx context.Context, fromID uint64, linkType, meta string) {} diff --git a/example/topo/full_linktype_topology_test.go b/example/topo/full_linktype_topology_test.go new file mode 100644 index 0000000..1c4b98f --- /dev/null +++ b/example/topo/full_linktype_topology_test.go @@ -0,0 +1,36 @@ +package topo + +import ( + "testing" +) + +func TestFullLinkTypeTopogoly(t *testing.T) { + // test topology of Neighbors Linktype + topoNeighbor := NewFullTopologyOfNeighbor(2) + + topoNeighbor.SetTaskID(0) + + n := topoNeighbor.GetNeighbors(0) + if len(n) != 2 { + t.Error() + } + if n[0] != 0 { + t.Error() + } + if n[1] != 1 { + t.Error() + } + + // test topology of Master Linktype + topoMaster := NewFullTopologyOfMaster(2) + + topoMaster.SetTaskID(0) + + m := topoMaster.GetNeighbors(0) + if len(m) != 1 { + t.Error() + } + if n[0] != 0 { + t.Error() + } +} diff --git a/example/topo/full_topo.go b/example/topo/full_topo.go index 1d10f76..7f32edb 100644 --- a/example/topo/full_topo.go +++ b/example/topo/full_topo.go @@ -1,60 +1,45 @@ package topo //The full structure is basically assume that every one is also parent for every else. -//And everyone else is communicating to get the data they need. +//And everyone else is communicating to get the data they need. Task 0 is forced/assumed to be the master. // //Also the star structure stays the same between epochs. type FullTopology struct { - numOfTasks uint64 - taskID uint64 - parents, children []uint64 + numOfTasks uint64 + taskID uint64 + all []uint64 } // TODO, do we really need to expose this? Ideally after proper construction of StarTopology // we should not need to set this again. func (t *FullTopology) SetTaskID(taskID uint64) { + t.all = make([]uint64, 0, t.numOfTasks) t.taskID = taskID - - // each task is parent of every else. - t.parents = make([]uint64, 0, t.numOfTasks-1) - for index := uint64(1); index < t.numOfTasks; index++ { - if index != t.taskID { - t.parents = append(t.parents, index) - } - } - - t.children = make([]uint64, 0, t.numOfTasks-1) - for index := uint64(1); index < t.numOfTasks; index++ { - if index != t.taskID { - t.children = append(t.children, index) - } + for index := uint64(0); index < t.numOfTasks; index++ { + t.all = append(t.all, index) } - } func (t *FullTopology) GetLinkTypes() []string { - return []string{"Parents", "Children"} + return []string{"Neighbors", "Master"} } -func (t *FullTopology) GetNeighbors(linkType string, epoch uint64) []uint64 { - res := make([]uint64, 0) - switch { - case linkType == "Parents": - res = t.parents - case linkType == "Children": - res = t.children +func (t *FullTopology) GetNeighbors(linkType string, epoch uint64) (res []uint64) { + switch linkType { + case "Neighbors": + res = t.all + case "Master": + res = []uint64{0} + default: + panic("") } - return res + return } -// TODO, do we really need to expose this? -func (t *FullTopology) SetNumberOfTasks(nt uint64) { t.numOfTasks = nt } - // Creates a new tree topology with given fanout and number of tasks. // This will be called during the task graph configuration. func NewFullTopology(nTasks uint64) *FullTopology { - m := &FullTopology{ + return &FullTopology{ numOfTasks: nTasks, } - return m } diff --git a/example/topo/full_topo_master.go b/example/topo/full_topo_master.go new file mode 100644 index 0000000..f19e649 --- /dev/null +++ b/example/topo/full_topo_master.go @@ -0,0 +1,34 @@ +package topo + +//The full structure is basically assume that every one is also parent for every else. +//And everyone else is communicating to get the data they need. Task 0 is forced/assumed to be the master. +// +//Also the star structure stays the same between epochs. +type FullTopologyOfMaster struct { + numOfTasks uint64 + taskID uint64 + all []uint64 +} + +// TODO, do we really need to expose this? Ideally after proper construction of StarTopology +// we should not need to set this again. +func (t *FullTopologyOfMaster) SetTaskID(taskID uint64) { + t.all = make([]uint64, 0, t.numOfTasks) + t.taskID = taskID + for index := uint64(0); index < t.numOfTasks; index++ { + t.all = append(t.all, index) + } +} + +func (t *FullTopologyOfMaster) GetNeighbors(epoch uint64) []uint64 { + res := []uint64{0} + return res +} + +// Creates a new tree topology with given fanout and number of tasks. +// This will be called during the task graph configuration. +func NewFullTopologyOfMaster(nTasks uint64) *FullTopologyOfMaster { + return &FullTopologyOfMaster{ + numOfTasks: nTasks, + } +} diff --git a/example/topo/full_topo_neighbor.go b/example/topo/full_topo_neighbor.go new file mode 100644 index 0000000..9bb9ebb --- /dev/null +++ b/example/topo/full_topo_neighbor.go @@ -0,0 +1,35 @@ +package topo + +//The full structure is basically assume that every one is also parent for every else. +//And everyone else is communicating to get the data they need. Task 0 is forced/assumed to be the master. +// +//Also the star structure stays the same between epochs. +type FullTopologyOfNegihbor struct { + numOfTasks uint64 + taskID uint64 + all []uint64 +} + +// TODO, do we really need to expose this? Ideally after proper construction of StarTopology +// we should not need to set this again. +func (t *FullTopologyOfNegihbor) SetTaskID(taskID uint64) { + t.all = make([]uint64, 0, t.numOfTasks) + t.taskID = taskID + for index := uint64(0); index < t.numOfTasks; index++ { + t.all = append(t.all, index) + } +} + +func (t *FullTopologyOfNegihbor) GetNeighbors(epoch uint64) []uint64 { + res := make([]uint64, 0) + res = t.all + return res +} + +// Creates a new tree topology with given fanout and number of tasks. +// This will be called during the task graph configuration. +func NewFullTopologyOfNeighbor(nTasks uint64) *FullTopologyOfNegihbor { + return &FullTopologyOfNegihbor{ + numOfTasks: nTasks, + } +} diff --git a/example/topo/full_topo_test.go b/example/topo/full_topo_test.go new file mode 100644 index 0000000..a536009 --- /dev/null +++ b/example/topo/full_topo_test.go @@ -0,0 +1,29 @@ +package topo + +import ( + "testing" +) + +func TestFullTopogoly(t *testing.T) { + topo := NewFullTopology(2) + + topo.SetTaskID(0) + + n := topo.GetNeighbors("Neighbors", 0) + if len(n) != 2 { + t.Error() + } + if n[0] != 0 { + t.Error() + } + if n[1] != 1 { + t.Error() + } + m := topo.GetNeighbors("Master", 0) + if len(m) != 1 { + t.Error() + } + if n[0] != 0 { + t.Error() + } +} diff --git a/example/topo/tree_linktype_topology_test.go b/example/topo/tree_linktype_topology_test.go new file mode 100644 index 0000000..a2ccbaf --- /dev/null +++ b/example/topo/tree_linktype_topology_test.go @@ -0,0 +1,88 @@ +package topo + +import "testing" + +type treeLinkTypeTopoTest struct { + id uint64 + parents, children []uint64 +} + +// 0 +// 1 2 +// 3 4 5 6 +// 7 +func TestTreeLinkTypeToplogy27(t *testing.T) { + tests27 := []treeLinkTypeTopoTest{ + { + uint64(0), + []uint64{}, []uint64{1, 2}, + }, + { + uint64(1), + []uint64{0}, []uint64{3, 4}, + }, + { + uint64(3), + []uint64{1}, []uint64{7}, + }, + { + uint64(5), + []uint64{2}, []uint64{}, + }, + } + testTreeLinkTypeTopology(2, 8, tests27, t) +} + +// 0 +// 1 2 +// 3 4 5 6 +// 7 8 +func TestTreeLinkTypeToplogy28(t *testing.T) { + tests28 := []treeLinkTypeTopoTest{ + { + uint64(0), + []uint64{}, []uint64{1, 2}, + }, + { + uint64(1), + []uint64{0}, []uint64{3, 4}, + }, + { + uint64(3), + []uint64{1}, []uint64{7, 8}, + }, + } + testTreeLinkTypeTopology(2, 9, tests28, t) +} + +func testTreeLinkTypeTopology(fanout, number uint64, tests []treeLinkTypeTopoTest, t *testing.T) { + for _, tt := range tests { + // For each row, we construct the tree topology and then + // set up the task id. + treeParentTopology := NewTreeTopologyOfParent(fanout, number) + treeParentTopology.SetTaskID(tt.id) + + parents := treeParentTopology.GetNeighbors(0) + if len(parents) != len(tt.parents) { + t.Errorf("TreeTopology27 got wrong number of parents for %q", tt.id) + } + for index, element := range parents { + if element != tt.parents[index] { + t.Errorf("Mismatch in %qth parent: expected %q got %q", index, element, tt.parents[index]) + } + } + + treeChildrenTopology := NewTreeTopologyOfChildren(fanout, number) + treeChildrenTopology.SetTaskID(tt.id) + + children := treeChildrenTopology.GetNeighbors(0) + if len(children) != len(tt.children) { + t.Errorf("TreeTopology27 got wrong number of children for %q", tt.id) + } + for index, element := range children { + if element != tt.children[index] { + t.Errorf("Mismatch in %qth children: expected %q got %q", index, element, tt.children[index]) + } + } + } +} diff --git a/example/topo/tree_topo_children.go b/example/topo/tree_topo_children.go new file mode 100644 index 0000000..f205bb5 --- /dev/null +++ b/example/topo/tree_topo_children.go @@ -0,0 +1,37 @@ +package topo + +//The tree structure is basically assume that all the task forms a tree. +//Also the tree structure stays the same between epochs. +type TreeTopologyOfChildren struct { + fanout, numOfTasks uint64 + taskID uint64 + children []uint64 +} + +func (t *TreeTopologyOfChildren) SetTaskID(taskID uint64) { + t.taskID = taskID + // Not the most efficient way to create parents and children, but + // since this is not on critical path, we are ok. + t.children = make([]uint64, 0, t.fanout) + + for index := uint64(1); index < t.numOfTasks; index++ { + parentID := (index - 1) / t.fanout + if parentID == taskID { + t.children = append(t.children, index) + } + } +} + +func (t *TreeTopologyOfChildren) GetNeighbors(epoch uint64) []uint64 { + return t.children +} + +// Creates a new tree topology with given fanout and number of tasks. +// This will be called during the task graph configuration. +func NewTreeTopologyOfChildren(fanout, nTasks uint64) *TreeTopologyOfChildren { + m := &TreeTopologyOfChildren{ + fanout: fanout, + numOfTasks: nTasks, + } + return m +} diff --git a/example/topo/tree_topo_parent.go b/example/topo/tree_topo_parent.go new file mode 100644 index 0000000..7a3460f --- /dev/null +++ b/example/topo/tree_topo_parent.go @@ -0,0 +1,40 @@ +package topo + +//The tree structure is basically assume that all the task forms a tree. +//Also the tree structure stays the same between epochs. +type TreeTopologyOfParent struct { + fanout, numOfTasks uint64 + taskID uint64 + parent []uint64 +} + +func (t *TreeTopologyOfParent) SetTaskID(taskID uint64) { + t.taskID = taskID + // Not the most efficient way to create parents and children, but + // since this is not on critical path, we are ok. + t.parent = make([]uint64, 0, 1) + + for index := uint64(1); index < t.numOfTasks; index++ { + parentID := (index - 1) / t.fanout + if index == taskID { + t.parent = append(t.parent, parentID) + if len(t.parent) > 1 { + panic("unexpcted number of partents for a tree topology") + } + } + } +} + +func (t *TreeTopologyOfParent) GetNeighbors(epoch uint64) []uint64 { + return t.parent +} + +// Creates a new tree topology with given fanout and number of tasks. +// This will be called during the task graph configuration. +func NewTreeTopologyOfParent(fanout, nTasks uint64) *TreeTopologyOfParent { + m := &TreeTopologyOfParent{ + fanout: fanout, + numOfTasks: nTasks, + } + return m +} diff --git a/example/topo/tree_topo_test.go b/example/topo/tree_topo_test.go index bc04be0..f7c9b11 100644 --- a/example/topo/tree_topo_test.go +++ b/example/topo/tree_topo_test.go @@ -62,7 +62,7 @@ func testTreeTopology(fanout, number uint64, tests []treeTopoTest, t *testing.T) treeTopology := NewTreeTopology(fanout, number) treeTopology.SetTaskID(tt.id) - linkTypes := treeTopology.GetLinkTypes(0) + linkTypes := treeTopology.GetLinkTypes() if len(linkTypes) != 2 { t.Error() } @@ -73,7 +73,7 @@ func testTreeTopology(fanout, number uint64, tests []treeTopoTest, t *testing.T) t.Error() } - parents := treeTopology.GetLinks("Parents", 0) + parents := treeTopology.GetNeighbors("Parents", 0) if len(parents) != len(tt.parents) { t.Errorf("TreeTopology27 got wrong number of parents for %q", tt.id) } @@ -83,7 +83,7 @@ func testTreeTopology(fanout, number uint64, tests []treeTopoTest, t *testing.T) } } - children := treeTopology.GetLinks("Children", 0) + children := treeTopology.GetNeighbors("Children", 0) if len(children) != len(tt.children) { t.Errorf("TreeTopology27 got wrong number of children for %q", tt.id) } diff --git a/filesystem/Readme.md b/filesystem/Readme.md new file mode 100644 index 0000000..05dc579 --- /dev/null +++ b/filesystem/Readme.md @@ -0,0 +1,39 @@ +# Azure Storage Client +This program provides filesystem interfaces that make it easy to access Azure Storage Client. + +# Installation +- Install Golang: https://golang.org/doc/install +- Get Azure SDK package: + +```sh +go get github.com/MSOpenTech/azure-sdk-for-go +``` +- Install: + +```sh +go install github.com/MSOpenTech/azure-sdk-for-go +``` + +# Deployment + +- follow http://azure.microsoft.com/en-gb/documentation/articles/storage-create-storage-account/ to create Azure Storage. +- Get your Azure Storage Accounts, Keys and BlobServiceBaseUrl. Like http://mystorageaccount.file.core.windows.net, the BlobServiceBaserUrl should be core.windows.net +- Use NewClient function, specify whether to use HTTPS, a specific REST API version + +Example: + +``` + import "github.com/taskgraph/taskgraph/filesystem" + ... + ... + cli, err := filesystem.NewAzureClient( + AzurestorageAccountName, + AzurestorageAccountKey, + "core.chinacloudapi.cn", + "2014-02-14", + true + ) +``` + +# License +[Apache 2.0](LICENSE-2.0.txt) \ No newline at end of file diff --git a/filesystem/azure.go b/filesystem/azure.go new file mode 100644 index 0000000..4c41a8f --- /dev/null +++ b/filesystem/azure.go @@ -0,0 +1,320 @@ +// TODO : updated the semantics of the Azure filesystem +// Explanation : +// current semantics is Container/Blob like "A/B", restricted by only one slash +// Might need to update the senmatics supported multiple slash +// (As same as the local system semantic) +// "/A/B/C/D", ignore the first slash, "A" represents the contianer name +// and "B/C/D" represents the Blob name. +// Correspendingly, the Blob function, Remove function, +// Exist function, Rename function need to change. + +package filesystem + +import ( + "bytes" + "encoding/base64" + "fmt" + "io" + "log" + "os" + "path" + "strings" + + "github.com/Azure/azure-sdk-for-go/storage" +) + +type AzureClient struct { + client *storage.Client + blobClient storage.BlobStorageClient +} + +type AzureFile struct { + path string + logger *log.Logger + client storage.BlobStorageClient +} + +// convertToAzurePath function +// convertToAzurePath splits the given name into two parts +// The first part represents the container's name, and the length of it shoulb be 32 due to Azure restriction +// The second part represents the blob's name +// It will return any error while converting +func convertToAzurePath(name string) (string, string, error) { + afterSplit := strings.Split(name, "/") + if len(afterSplit[0]) != 32 { + return "", "", fmt.Errorf("azureClient : the length of container should be 32") + } + blobName := "" + if len(afterSplit) > 1 { + blobName = name[len(afterSplit[0])+1:] + } + return afterSplit[0], blobName, nil +} + +//AzureClient -> Delete function +// Delete specific Blob for input path +func (c *AzureClient) Remove(name string) error { + afterSplit := strings.Split(name, "/") + if len(afterSplit) == 1 && len(afterSplit[0]) == 32 { + _, err := c.blobClient.DeleteContainerIfExists(name) + if err != nil { + return err + } + return nil + } + containerName, blobName, err := convertToAzurePath(name) + if err != nil { + return err + } + _, err = c.blobClient.DeleteBlobIfExists(containerName, blobName) + return err + +} + +// AzureClient -> Exist function +// support check the contianer or blob if exist or not +func (c *AzureClient) Exists(name string) (bool, error) { + containerName, blobName, err := convertToAzurePath(name) + if err != nil { + return false, err + } + if blobName != "" { + return c.blobClient.BlobExists(containerName, blobName) + } else { + return c.blobClient.ContainerExists(containerName) + } + +} + +// Azure prevent user renaming their blob +// Thus this function firstly copy the source blob, +// when finished, delete the source blob. +// http://stackoverflow.com/questions/3734672/azure-storage-blob-rename +func (c *AzureClient) moveBlob(dstContainerName, dstBlobName, srcContainerName, srcBlobName string, isContainerRename bool) error { + dstBlobUrl := c.blobClient.GetBlobURL(dstContainerName, dstBlobName) + srcBlobUrl := c.blobClient.GetBlobURL(srcContainerName, srcBlobName) + if dstBlobUrl != srcBlobUrl { + err := c.blobClient.CopyBlob(dstContainerName, dstBlobName, srcBlobUrl) + if err != nil { + return err + } + if !isContainerRename { + err = c.blobClient.DeleteBlob(srcContainerName, srcBlobName) + if err != nil { + return err + } + } + } + return nil +} + +// AzureClient -> Rename function +// support user renaming contianer/blob +func (c *AzureClient) Rename(oldpath, newpath string) error { + exist, err := c.Exists(oldpath) + if err != nil { + return err + } + if !exist { + return fmt.Errorf("azureClient : oldpath does not exist") + } + srcContainerName, srcBlobName, err := convertToAzurePath(oldpath) + if err != nil { + return err + } + dstContainerName, dstBlobName, err := convertToAzurePath(newpath) + if err != nil { + return err + } + if srcBlobName == "" && dstBlobName == "" { + resp, err := c.blobClient.ListBlobs(srcContainerName, storage.ListBlobsParameters{Marker: ""}) + if err != nil { + return err + } + _, err = c.blobClient.CreateContainerIfNotExists(dstContainerName, storage.ContainerAccessTypeBlob) + if err != nil { + return err + } + + for _, blob := range resp.Blobs { + err = c.moveBlob(dstContainerName, blob.Name, srcContainerName, blob.Name, true) + if err != nil { + return err + } + } + err = c.blobClient.DeleteContainer(srcContainerName) + if err != nil { + return err + } + } else if srcBlobName != "" && dstBlobName != "" { + c.moveBlob(dstContainerName, dstBlobName, srcContainerName, srcBlobName, false) + } else { + return fmt.Errorf("Rename path does not match") + } + + return nil +} + +// AzureClient -> OpenReadCloser function +// implement by the providing function +func (c *AzureClient) OpenReadCloser(name string) (io.ReadCloser, error) { + containerName, blobName, err := convertToAzurePath(name) + if err != nil { + return nil, err + } + return c.blobClient.GetBlob(containerName, blobName) +} + +//AzureClient -> OpenWriteCloser function +// If not exist, Create corresponding Container and blob. +func (c *AzureClient) OpenWriteCloser(name string) (io.WriteCloser, error) { + exist, err := c.Exists(name) + if err != nil { + return nil, err + } + containerName, blobName, err := convertToAzurePath(name) + if err != nil { + return nil, err + } + if !exist { + _, err = c.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + return nil, err + } + err = c.blobClient.CreateBlockBlob(containerName, blobName) + if err != nil { + return nil, err + } + } + return &AzureFile{ + path: name, + logger: log.New(os.Stdout, "", log.Lshortfile|log.LstdFlags), + client: c.blobClient, + }, nil +} + +// Follow io.Writer rules +// use BlockList mechanism of Azure Blob to achieve this interface +// BlockList divided block into committed block and uncommitted ones +// For every write op of user, +// it create new block list with old commited block and user write content, +// and commit this one to modifies the blob commited block list +// If failed in the process, it would write nothing to blob +// return 0, error to user +func (f *AzureFile) Write(b []byte) (int, error) { + if len(b) == 0 { + return 0, fmt.Errorf("Need write content") + } + cnt, blob, err := convertToAzurePath(f.path) + if err != nil { + return 0, err + } + + blockList, err := f.client.GetBlockList(cnt, blob, storage.BlockListTypeAll) + if err != nil { + return 0, err + } + + // blockLen is a naming rule for block, + // uses base64.StdEncoding.EncodeToString(fmt.Sprintf("%011d\n", blocksLen))) + + // blockLen initially set to committed block size + // if exist uncommitted block of same name, + // it will rewrite the uncommitted block content + blocksLen := len(blockList.CommittedBlocks) + amendList := []storage.Block{} + for _, v := range blockList.CommittedBlocks { + amendList = append(amendList, storage.Block{v.Name, storage.BlockStatusCommitted}) + } + + var chunkSize int = storage.MaxBlobBlockSize + inputSourceReader := bytes.NewReader(b) + chunk := make([]byte, chunkSize) + for { + n, err := inputSourceReader.Read(chunk) + if err == io.EOF { + break + } + if err != nil { + return 0, err + } + + blockId := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%011d\n", blocksLen))) + data := chunk[:n] + err = f.client.PutBlock(cnt, blob, blockId, data) + if err != nil { + return 0, err + } + // Add current uncommitted block to temporary block list + amendList = append(amendList, storage.Block{blockId, storage.BlockStatusUncommitted}) + blocksLen++ + } + // update block list ot blob committed block list. + err = f.client.PutBlockList(cnt, blob, amendList) + if err != nil { + return 0, err + } + return len(b), nil +} + +func (f *AzureFile) Close() error { + return nil +} + +// AzureClient -> Glob function +// only supports '*', '?' +// Syntax: +// cntName?/part.* +func (c *AzureClient) Glob(pattern string) (matches []string, err error) { + afterSplit := strings.Split(pattern, "/") + cntPattern, blobPattern := afterSplit[0], afterSplit[1] + if len(afterSplit) != 2 { + return nil, fmt.Errorf("Glob pattern should follow the Syntax") + } + resp, err := c.blobClient.ListContainers(storage.ListContainersParameters{Prefix: ""}) + if err != nil { + return nil, err + } + for _, cnt := range resp.Containers { + matched, err := path.Match(cntPattern, cnt.Name) + if err != nil { + return nil, err + } + if !matched { + continue + } + resp, err := c.blobClient.ListBlobs(cnt.Name, storage.ListBlobsParameters{Marker: ""}) + if err != nil { + return nil, err + } + for _, v := range resp.Blobs { + matched, err := path.Match(blobPattern, v.Name) + if err != nil { + return nil, err + } + if matched { + matches = append(matches, cnt.Name+"/"+v.Name) + } + } + } + return matches, nil +} + +// NewAzureClient function +// NewClient constructs a StorageClient and blobStorageClinet. +// This should be used if the caller wants to specify +// whether to use HTTPS, a specific REST API version or a +// custom storage endpoint than Azure Public Cloud. +// Recommended API version "2014-02-14" +// synax : +// AzurestorageAccountName, AzurestorageAccountKey, "core.chinacloudapi.cn", "2014-02-14", true +func NewAzureClient(accountName, accountKey, blobServiceBaseUrl, apiVersion string, useHttps bool) (*AzureClient, error) { + cli, err := storage.NewClient(accountName, accountKey, blobServiceBaseUrl, apiVersion, useHttps) + if err != nil { + return nil, err + } + return &AzureClient{ + client: &cli, + blobClient: cli.GetBlobService(), + }, nil +} diff --git a/filesystem/azure_test.go b/filesystem/azure_test.go new file mode 100644 index 0000000..8946837 --- /dev/null +++ b/filesystem/azure_test.go @@ -0,0 +1,377 @@ +package filesystem + +import ( + "bytes" + "crypto/rand" + "fmt" + "io/ioutil" + "os" + "strconv" + "testing" + + "github.com/Azure/azure-sdk-for-go/storage" +) + +// If testing, user should add at least AzureStorageAccountName, +// AzureStorageAccountKey, AzureBlobServiceBaseUrl to env. +// Azure Client setting example : +// { +// TestAzureAccountName : yourAccountName +// TestAzureAccountKey : yourKey +// TestAzureBlobServiceBaseUrl : "core.chinacloudapi.cn" +// } + +func TestAzureClientWriteAndReadCloser(t *testing.T) { + type nodeIO struct { + inputString []string + } + + // import test cases + // add input content to check correctness of + // azure storage client read/write modules + var flagtests = []nodeIO{ + {[]string{"some data", "some data", "some data"}}, + {[]string{"a", "b", "c"}}, + } + str := "" + for i := 0; i < 2000; i++ { + str = str + strconv.Itoa(i) + } + flagtests = append(flagtests, nodeIO{[]string{str}}) + + // testing + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(containerName) + + for i, tt := range flagtests { + blobName := "testcase" + strconv.Itoa(i) + writeCloser, err := cli.OpenWriteCloser(containerName + "/" + blobName) + if err != nil { + t.Fatalf("OpenWriteCloser failed: %v", err) + } + defer cli.blobClient.DeleteBlob(containerName, blobName) + var cmp []byte + for _, content := range tt.inputString { + data := []byte(content) + writeLen, err := writeCloser.Write(data) + if err != nil { + t.Fatalf("Write failed: %v", err) + } + if writeLen != len(data) { + t.Fatalf("Write num is not correct. Get = %d, Want = %d", writeLen, len(data)) + } + cmp = append(cmp, data...) + } + writeCloser.Close() + + readCloser, err := cli.OpenReadCloser(containerName + "/" + blobName) + if err != nil { + t.Fatalf("OpenReadCloser failed: %v", err) + } + b, err := ioutil.ReadAll(readCloser) + if err != nil { + t.Fatalf("Read failed: %v", err) + } + readCloser.Close() + + if bytes.Compare(b, cmp) != 0 { + t.Fatalf("Read result isn't correct. Get = %s, Want = %s", string(b), string(cmp)) + } + } + +} + +func TestAzureClientExistContainer(t *testing.T) { + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + blobName, err := randString(5) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + ok, err := cli.Exists(containerName) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Errorf("Existing contianer returned as non-existing: %s/%s", containerName, blobName) + } + + err = cli.blobClient.DeleteContainer(containerName) + if err != nil { + t.Fatal(err) + } + ok, err = cli.Exists(containerName) + if err != nil { + t.Fatal(err) + } + if ok { + t.Errorf("Non-existing contianer returned as existing: %s/%s", containerName, blobName) + } +} + +func TestAzureClientExistBlob(t *testing.T) { + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + blobName, err := randString(5) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(containerName) + err = cli.blobClient.CreateBlockBlob(containerName, blobName) + if err != nil { + t.Fatal(err) + } + + defer cli.blobClient.DeleteBlob(containerName, blobName) + ok, err := cli.Exists(containerName + "/" + blobName + ".foo") + if err != nil { + t.Fatal(err) + } + if ok { + t.Errorf("Non-existing blob returned as existing: %s/%s", containerName, blobName) + } + ok, err = cli.Exists(containerName + "/" + blobName) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatalf("!Existing blob returned as non-existing: %s/%s", containerName, blobName) + } +} + +func TestAzureClientRemoveBlob(t *testing.T) { + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + blobName, err := randString(5) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(containerName) + err = cli.blobClient.CreateBlockBlob(containerName, blobName) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteBlob(containerName, blobName) + + cli.Remove(containerName + "/" + blobName) + exist, err := cli.Exists(containerName + "/" + blobName) + if err != nil { + t.Fatal(err) + } + if exist { + t.Fatalf("Pointed blob removed failed") + } +} + +func TestAzureClientRemoveContainer(t *testing.T) { + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(containerName) + + cli.Remove(containerName) + exist, err := cli.Exists(containerName) + if err != nil { + t.Fatal(err) + } + if exist { + t.Fatalf("Pointed contianer removed failed") + } +} + +func TestAzureClientGlob(t *testing.T) { + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + blobName, err := randString(5) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(containerName) + err = cli.blobClient.CreateBlockBlob(containerName, "1") + if err != nil { + t.Fatal(err) + } + + defer cli.blobClient.DeleteBlob(containerName, blobName) + err = cli.blobClient.CreateBlockBlob(containerName, "1.txt") + if err != nil { + t.Fatal(err) + } + + defer cli.blobClient.DeleteBlob(containerName, blobName) + err = cli.blobClient.CreateBlockBlob(containerName, "2.txt") + if err != nil { + t.Fatal(err) + } + + defer cli.blobClient.DeleteBlob(containerName, blobName) + globPath := containerName + "/*.txt" + names, err := cli.Glob(globPath) + if err != nil { + t.Fatalf("Glob(%s) failed: %v", globPath, err) + } + // make sure glob result includes all *.txt files + nameMap := make(map[string]int) + for _, name := range names { + nameMap[name] += 1 + } + if len(names) != 2 || + nameMap[containerName+"/1.txt"] != 1 || nameMap[containerName+"/2.txt"] != 1 { + t.Fatalf("Glob result isn't correct. Get = %v, Want = %v", nameMap, []string{"/tmp/testing/1.txt", "/tmp/testing/2.txt"}) + } +} + +func TestAzureClientRenameBlob(t *testing.T) { + cli := setupAzureTest(t) + containerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + blobName, err := randString(5) + if err != nil { + t.Fatal(err) + } + _, err = cli.blobClient.CreateContainerIfNotExists(containerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(containerName) + err = cli.blobClient.CreateBlockBlob(containerName, blobName) + if err != nil { + t.Fatal(err) + } + cli.Rename(containerName+"/"+blobName, containerName+"/"+blobName+"-Rename") + exist, _ := cli.Exists(containerName + "/" + blobName + "-Rename") + if !exist { + t.Fatalf("Rename failed") + } + exist, err = cli.Exists(containerName + "/" + blobName) + if err != nil { + t.Fatal(err) + } + if exist { + t.Fatalf("Rename failed") + } + defer cli.blobClient.DeleteBlob(containerName, blobName+"-Rename") +} + +func TestAzureClientRenameContainer(t *testing.T) { + cli := setupAzureTest(t) + srcContainerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + + dstContainerName, err := randString(32) + if err != nil { + t.Fatal(err) + } + blobName := "rename" + _, err = cli.blobClient.CreateContainerIfNotExists(srcContainerName, storage.ContainerAccessTypeBlob) + if err != nil { + t.Fatal(err) + } + defer cli.blobClient.DeleteContainer(srcContainerName) + err = cli.blobClient.CreateBlockBlob(srcContainerName, blobName+"01") + if err != nil { + t.Fatal(err) + } + err = cli.blobClient.CreateBlockBlob(srcContainerName, blobName+"02") + if err != nil { + t.Fatal(err) + } + + cli.Rename(srcContainerName, dstContainerName) + exist, _ := cli.Exists(dstContainerName + "/" + blobName + "01") + if !exist { + t.Fatalf("Rename failed") + } + exist, err = cli.Exists(dstContainerName + "/" + blobName + "02") + if err != nil { + t.Fatal(err) + } + if !exist { + t.Fatalf("Rename failed") + } + + exist, err = cli.Exists(srcContainerName) + if err != nil { + t.Fatal(err) + } + if exist { + t.Fatalf("Rename failed") + } + defer cli.blobClient.DeleteContainer(dstContainerName) +} + +func setupAzureTest(t *testing.T) *AzureClient { + TestAzureAccountName := os.Getenv("TestAzureAccountName") + TestAzureAccountKey := os.Getenv("TestAzureAccountKey") + TestAzureBlobServiceBaseUrl := os.Getenv("TestAzureBlobServiceBaseUrl") + apiVersion := "2014-02-14" + useHttps := true + if TestAzureAccountName == "" || TestAzureAccountKey == "" || TestAzureBlobServiceBaseUrl == "" { + t.Skip("Azure config not specified.") + } + client, err := NewAzureClient(TestAzureAccountName, TestAzureAccountKey, TestAzureBlobServiceBaseUrl, apiVersion, useHttps) + if err != nil { + t.Fatalf("NewAzureClient(%s, %s, %s) failed: %v", + TestAzureAccountName, TestAzureAccountKey, TestAzureBlobServiceBaseUrl, err) + } + return client +} + +func randString(n int) (string, error) { + if n <= 0 { + return "", fmt.Errorf("negative number") + } + const alphanum = "0123456789abcdefghijklmnopqrstuvwxyz" + var bytes = make([]byte, n) + rand.Read(bytes) + for i, b := range bytes { + bytes[i] = alphanum[b%byte(len(alphanum))] + } + return string(bytes), nil +} diff --git a/filesystem/client_interface.go b/filesystem/client_interface.go index 69a3b6c..580b905 100644 --- a/filesystem/client_interface.go +++ b/filesystem/client_interface.go @@ -13,4 +13,6 @@ type Client interface { // Glob returns the names of all files matching pattern or // nil if there is no matching file. Glob(pattern string) (matches []string, err error) + // Remove specific file in filesystem + Remove(name string) error } diff --git a/filesystem/hdfs.go b/filesystem/hdfs.go index 745eb8e..cb04fa8 100644 --- a/filesystem/hdfs.go +++ b/filesystem/hdfs.go @@ -12,6 +12,7 @@ import ( "os" "path" "strings" + "time" "github.com/colinmarc/hdfs" ) @@ -27,17 +28,15 @@ type hdfsConfig struct { } type HdfsClient struct { - client *hdfs.Client hdfsConfig } func NewHdfsClient(namenodeAddr, webHdfsAddr, user string) (Client, error) { - client, err := hdfs.NewForUser(namenodeAddr, user) + _, err := hdfs.NewForUser(namenodeAddr, user) if err != nil { return nil, err } return &HdfsClient{ - client: client, hdfsConfig: hdfsConfig{ namenodeAddr: namenodeAddr, webHdfsAddr: webHdfsAddr, @@ -46,8 +45,20 @@ func NewHdfsClient(namenodeAddr, webHdfsAddr, user string) (Client, error) { }, nil } +func (c *HdfsClient) Remove(name string) error { + client, err := hdfs.NewForUser(c.hdfsConfig.namenodeAddr, c.hdfsConfig.user) + if err != nil { + return err + } + return client.Remove(name) +} + func (c *HdfsClient) OpenReadCloser(name string) (io.ReadCloser, error) { - return c.client.Open(name) + client, err := hdfs.NewForUser(c.hdfsConfig.namenodeAddr, c.hdfsConfig.user) + if err != nil { + return nil, err + } + return client.Open(name) } func (c *HdfsClient) OpenWriteCloser(name string) (io.WriteCloser, error) { @@ -56,7 +67,11 @@ func (c *HdfsClient) OpenWriteCloser(name string) (io.WriteCloser, error) { return nil, err } if !exist { - err := c.client.CreateEmptyFile(name) + client, err := hdfs.NewForUser(c.hdfsConfig.namenodeAddr, c.hdfsConfig.user) + if err != nil { + return nil, err + } + err = client.CreateEmptyFile(name) if err != nil { return nil, err } @@ -69,12 +84,20 @@ func (c *HdfsClient) OpenWriteCloser(name string) (io.WriteCloser, error) { } func (c *HdfsClient) Exists(name string) (bool, error) { - _, err := c.client.Stat(name) + client, err := hdfs.NewForUser(c.hdfsConfig.namenodeAddr, c.hdfsConfig.user) + if err != nil { + return false, err + } + _, err = client.Stat(name) return existCommon(err) } func (c *HdfsClient) Rename(oldpath, newpath string) error { - return c.client.Rename(oldpath, newpath) + client, err := hdfs.NewForUser(c.hdfsConfig.namenodeAddr, c.hdfsConfig.user) + if err != nil { + return err + } + return client.Rename(oldpath, newpath) } // only supports '*', '?' @@ -102,10 +125,15 @@ func (c *HdfsClient) Glob(pattern string) (matches []string, err error) { } func (c *HdfsClient) glob(dir string, names []string) (m []string, err error) { + + client, err := hdfs.NewForUser(c.hdfsConfig.namenodeAddr, c.hdfsConfig.user) + if err != nil { + return nil, err + } name := names[0] var dirs []string if hasMeta(name) { - fileInfos, err := c.client.ReadDir(dir) + fileInfos, err := client.ReadDir(dir) if err != nil { return nil, err } @@ -162,12 +190,22 @@ func (f *HdfsFile) Write(b []byte) (int, error) { return 0, fmt.Errorf("Write: NewRequest failed: %v", err) } // no redirect - resp, err := tr.RoundTrip(req) - if err != nil { - return 0, fmt.Errorf("Write: RoundTrip failed: %v", err) + + var resp *http.Response + var loc string + for retry := 0; retry < 3 && (err != nil || loc == ""); retry++ { + time.Sleep(300 * time.Millisecond) + resp, err := tr.RoundTrip(req) + if err != nil { + continue + } + defer resp.Body.Close() + loc = resp.Header.Get("Location") + } + + if loc == "" { + return 0, fmt.Errorf("Write: Failed retrieving datanode location.") } - defer resp.Body.Close() - loc := resp.Header.Get("Location") u, err := url.ParseRequestURI(loc) if err != nil { @@ -175,6 +213,10 @@ func (f *HdfsFile) Write(b []byte) (int, error) { } // POST request to datanode. resp, err = http.Post(u.String(), "application/octet-stream", bytes.NewBuffer(b)) + for retry := 0; retry < 3 && err != nil; retry++ { + time.Sleep(1 * time.Second) + resp, err = http.Post(u.String(), "application/octet-stream", bytes.NewBuffer(b)) + } if err != nil { return 0, fmt.Errorf("Write: POST to datanode (%s) failed: %v", u.String(), err) } diff --git a/filesystem/hdfs_test.go b/filesystem/hdfs_test.go index 3a0b9e7..dabce4f 100644 --- a/filesystem/hdfs_test.go +++ b/filesystem/hdfs_test.go @@ -5,6 +5,8 @@ import ( "io/ioutil" "os" "testing" + + "github.com/colinmarc/hdfs" ) var ( @@ -23,7 +25,7 @@ func init() { // to clean up files or dirs manually. func TestHdfsClientWrite(t *testing.T) { - client := setupHdfsTest(t) + client, c := setupHdfsTest(t) writeCloser, err := client.OpenWriteCloser("/tmp/testing") if err != nil { t.Fatalf("OpenWriteCloser failed: %v", err) @@ -48,14 +50,12 @@ func TestHdfsClientWrite(t *testing.T) { t.Fatalf("Read result isn't correct. Get = %s, Want = %s", string(b), string(data)) } - c := client.(*HdfsClient).client c.Remove("/tmp/testing") } func TestHdfsClientGlob(t *testing.T) { - client := setupHdfsTest(t) + client, c := setupHdfsTest(t) - c := client.(*HdfsClient).client c.Mkdir("/tmp/testing", 0644) c.CreateEmptyFile("/tmp/testing/1") c.CreateEmptyFile("/tmp/testing/1.txt") @@ -79,9 +79,8 @@ func TestHdfsClientGlob(t *testing.T) { } func TestHdfsClientRename(t *testing.T) { - client := setupHdfsTest(t) + client, c := setupHdfsTest(t) - c := client.(*HdfsClient).client c.CreateEmptyFile("/tmp/testing") client.Rename("/tmp/testing", "/tmp/tesing-renamed") @@ -92,7 +91,7 @@ func TestHdfsClientRename(t *testing.T) { c.Remove("/tmp/renamed") } -func setupHdfsTest(t *testing.T) Client { +func setupHdfsTest(t *testing.T) (Client, *hdfs.Client) { if namenodeAddr == "" || webhdfsAddr == "" || hdfsUser == "" { t.Skip("HDFS config not specified.") } @@ -101,5 +100,10 @@ func setupHdfsTest(t *testing.T) Client { t.Fatalf("NewHdfsClient(%s, %s) failed: %v", namenodeAddr, webhdfsAddr, err) } - return client + hClient, err := hdfs.NewForUser(namenodeAddr, hdfsUser) + if err != nil { + t.Fatalf("hdfs.NewForUser(%s, %s) failed: %v", + namenodeAddr, webhdfsAddr, err) + } + return client, hClient } diff --git a/filesystem/local_fs.go b/filesystem/local_fs.go index a4963fb..774e3d5 100644 --- a/filesystem/local_fs.go +++ b/filesystem/local_fs.go @@ -29,7 +29,7 @@ func (c *localFSClient) OpenWriteCloser(name string) (io.WriteCloser, error) { } return f, nil } - return os.Open(name) + return os.OpenFile(name, os.O_WRONLY, 0) } func (c *localFSClient) Exists(name string) (bool, error) { @@ -45,6 +45,10 @@ func (c *localFSClient) Glob(pattern string) (matches []string, err error) { return filepath.Glob(pattern) } +func (c *localFSClient) Remove(name string) error { + return os.Remove(name) +} + func existCommon(err error) (bool, error) { if err == nil { return true, nil diff --git a/framework/bootstrap.go b/framework/bootstrap.go index 18b565a..8714c02 100644 --- a/framework/bootstrap.go +++ b/framework/bootstrap.go @@ -24,9 +24,17 @@ func NewBootStrap(jobName string, etcdURLs []string, ln net.Listener, logger *lo } } -func (f *framework) SetTaskBuilder(taskBuilder taskgraph.TaskBuilder) { f.taskBuilder = taskBuilder } +func (f *framework) SetTaskBuilder(taskBuilder taskgraph.TaskBuilder) { + f.taskBuilder = taskBuilder +} -func (f *framework) SetTopology(topology taskgraph.Topology) { f.topology = topology } +//The user add their topology link to the original topology by this function +func (f *framework) AddLinkage(linkType string, topology taskgraph.Topology) { + if f.topology == nil { + f.topology = make(map[string]taskgraph.Topology) + } + f.topology[linkType] = topology +} func (f *framework) Start() { var err error @@ -43,16 +51,16 @@ func (f *framework) Start() { f.log.SetPrefix(fmt.Sprintf("task %d: ", f.taskID)) - f.epochChan = make(chan uint64, 1) // grab epoch from etcd - f.epochStop = make(chan bool, 1) // stop etcd watch + f.epochWatcher = make(chan uint64, 1) // grab epoch from etcd + f.epochWatchStop = make(chan bool, 1) // stop etcd watch // meta will have epoch prepended so we must get epoch before any watch on meta - f.epoch, err = etcdutil.GetAndWatchEpoch(f.etcdClient, f.name, f.epochChan, f.epochStop) + f.epoch, err = etcdutil.GetAndWatchEpoch(f.etcdClient, f.name, f.epochWatcher, f.epochWatchStop) if err != nil { f.log.Fatalf("WatchEpoch failed: %v", err) } if f.epoch == exitEpoch { f.log.Printf("found that job has finished\n") - f.epochStop <- true + f.epochWatchStop <- true return } f.log.Printf("starting at epoch %d\n", f.epoch) @@ -61,37 +69,39 @@ func (f *framework) Start() { // Both should be initialized at this point. // Get the task implementation and topology for this node (indentified by taskID) f.task = f.taskBuilder.GetTask(f.taskID) - f.topology.SetTaskID(f.taskID) + + // SetTaskID to every linkType topology in our map topology container + for key, _ := range f.topology { + f.topology[key].SetTaskID(f.taskID) + } f.heartbeat() - f.setupChannels() + f.setup() f.task.Init(f.taskID, f) f.run() f.releaseResource() f.task.Exit() } -func (f *framework) setupChannels() { - f.httpStop = make(chan struct{}) - f.metaChan = make(chan *metaChange, 100) - f.dataReqtoSendChan = make(chan *dataRequest, 100) - f.dataRespChan = make(chan *dataResponse, 100) - f.epochCheckChan = make(chan *epochCheck, 100) +func (f *framework) setup() { + f.globalStop = make(chan struct{}) + f.metaChan = make(chan *metaChange, 1) + f.dataReqtoSendChan = make(chan *dataRequest, 1) + f.dataRespChan = make(chan *dataResponse, 1) + f.epochCheckChan = make(chan *epochCheck, 1) } func (f *framework) run() { f.log.Printf("framework starts to run") defer f.log.Printf("framework stops running.") f.setEpochStarted() - // We need put off starting server here after task#SetEpoch is called. go f.startHTTP() - // this for-select is primarily used to synchronize epoch specific events. + // Central event handling for { select { - case nextEpoch, ok := <-f.epochChan: + case nextEpoch, ok := <-f.epochWatcher: f.releaseEpochResource() - if !ok { // single task exit - nextEpoch = exitEpoch + if !ok { // task is killed return } f.epoch = nextEpoch @@ -108,19 +118,19 @@ func (f *framework) run() { // the epoch that was meant for this event. This context will be passed // to user event handler functions and used to ask framework to do work later // with previous information. - f.handleMetaChange(f.createContext(), meta.from, meta.who, meta.meta) + go f.handleMetaChange(f.userCtx, meta.from, meta.linkType, meta.meta) case req := <-f.dataReqtoSendChan: if req.epoch != f.epoch { - f.log.Printf("epoch mismatch: req-to-send epoch: %d, current epoch: %d", req.epoch, f.epoch) + f.log.Printf("abort data request, to %d, epoch %d, method %s", req.taskID, req.epoch, req.method) break } go f.sendRequest(req) case resp := <-f.dataRespChan: if resp.epoch != f.epoch { - f.log.Printf("epoch mismatch: response epoch: %d, current epoch: %d", resp.epoch, f.epoch) + f.log.Printf("abort data response, to %d, epoch: %d, method %d", resp.taskID, resp.epoch, resp.method) break } - f.handleDataResp(f.createContext(), resp) + f.handleDataResp(f.userCtx, resp) case ec := <-f.epochCheckChan: if ec.epoch != f.epoch { ec.fail() @@ -132,34 +142,29 @@ func (f *framework) run() { } func (f *framework) setEpochStarted() { - f.epochPassed = make(chan struct{}) // Each epoch have a new meta map f.metaNotified = make(map[string]bool) - f.task.SetEpoch(f.createContext(), f.epoch) + f.userCtx = context.WithValue(context.Background(), epochKey, f.epoch) + f.userCtx, f.userCtxCancel = context.WithCancel(f.userCtx) + + f.task.EnterEpoch(f.userCtx, f.epoch) + // setup etcd watches - // - create self's parent and child meta flag - // - watch parents' child meta flag - // - watch children's parent meta flag - for _, linkType := range f.topology.GetLinkTypes() { - f.watchMeta(linkType, f.topology.GetNeighbors(linkType, f.epoch)) - } + f.watchMeta() } func (f *framework) releaseEpochResource() { - close(f.epochPassed) - for _, c := range f.metaStops { - c <- true - } - f.metaStops = nil + f.userCtxCancel() + f.metaWatchStop <- true } // release resources: heartbeat, epoch watch. func (f *framework) releaseResource() { f.log.Printf("framework is releasing resources...\n") - f.epochStop <- true - close(f.heartbeatStop) - f.stopHTTP() + f.epochWatchStop <- true + close(f.globalStop) + f.ln.Close() // stop grpc server } // occupyTask will grab the first unassigned task and register itself on etcd. @@ -170,7 +175,10 @@ func (f *framework) occupyTask() error { return err } f.log.Printf("standby grabbed free task %d", freeTask) - ok := etcdutil.TryOccupyTask(f.etcdClient, f.name, freeTask, f.ln.Addr().String()) + ok, err := etcdutil.TryOccupyTask(f.etcdClient, f.name, freeTask, f.ln.Addr().String()) + if err != nil { + return err + } if ok { f.taskID = freeTask return nil @@ -179,59 +187,58 @@ func (f *framework) occupyTask() error { } } -func (f *framework) watchMeta(linkType string, taskIDs []uint64) { - stops := make([]chan bool, len(taskIDs)) - - for i, taskID := range taskIDs { - stop := make(chan bool, 1) - stops[i] = stop - - watchPath := etcdutil.MetaPath(linkType, f.name, taskID) +func parseMetaValue(val string) (epoch, taskID uint64, linkType, meta string, err error) { + values := strings.SplitN(val, "-", 4) + // epoch is prepended to meta. When a new one starts and replaces + // the old one, it doesn't need to handle previous things, whose + // epoch is smaller than current one. + epoch, err = strconv.ParseUint(values[0], 10, 64) + if err != nil { + err = fmt.Errorf("not an unit64 epoch prepended: %s", values[0]) + return + } + taskID, err = strconv.ParseUint(values[1], 10, 64) + if err != nil { + err = fmt.Errorf("not an unit64 taskID prepended: %s", values[1]) + return + } + linkType = values[2] + meta = values[3] + return +} - // When a node working for a task crashed, a new node will take over - // the task and continue what's left. It assumes that progress is stalled - // until the new node comes (i.e. epoch won't change). - responseHandler := func(resp *etcd.Response, taskID uint64) { - if resp.Action != "set" && resp.Action != "get" { - return - } - // epoch is prepended to meta. When a new one starts and replaces - // the old one, it doesn't need to handle previous things, whose - // epoch is smaller than current one. - values := strings.SplitN(resp.Node.Value, "-", 2) - ep, err := strconv.ParseUint(values[0], 10, 64) - if err != nil { - f.log.Panicf("WARN: not a unit64 prepended to meta: %s", values[0]) - } - f.metaChan <- &metaChange{ - from: taskID, - who: linkType, - epoch: ep, - meta: values[1], - } +// watch meta flag and receive any notification. +func (f *framework) watchMeta() { + watchPath := etcdutil.MetaPath(f.name, f.taskID) + + // The function parses the watch response of meta flag, and passes the result + // to central event handling. + responseHandler := func(resp *etcd.Response) { + // Note: Why "get"? + // We first try to get the index. If there's value in it, we also parse it. + if resp.Action != "set" && resp.Action != "get" { + return } - - // Need to pass in taskID to make it work. Didn't know why. - err := etcdutil.WatchMeta(f.etcdClient, taskID, watchPath, stop, responseHandler) + epoch, taskID, linkType, meta, err := parseMetaValue(resp.Node.Value) if err != nil { - f.log.Panicf("WatchMeta failed. path: %s, err: %v", watchPath, err) + f.log.Panicf("parseMetaValue failed: %v", err) + } + f.metaChan <- &metaChange{ + from: taskID, + epoch: epoch, + linkType: linkType, + meta: meta, } } - f.metaStops = append(f.metaStops, stops...) -} -func (f *framework) handleMetaChange(ctx context.Context, taskID uint64, linkType, meta string) { - // check if meta is handled before. - tm := taskMeta(taskID, meta) - if _, ok := f.metaNotified[tm]; ok { - return + f.metaWatchStop = make(chan bool, 1) + // Need to pass in taskID to make it work. Didn't know why. + err := etcdutil.WatchMeta(f.etcdClient, watchPath, f.metaWatchStop, responseHandler) + if err != nil { + f.log.Panicf("WatchMeta failed. path: %s, err: %v", watchPath, err) } - f.metaNotified[tm] = true - - f.task.MetaReady(ctx, taskID, linkType, meta) - } -func taskMeta(taskID uint64, meta string) string { - return fmt.Sprintf("%s-%s", strconv.FormatUint(taskID, 10), meta) +func (f *framework) handleMetaChange(ctx context.Context, taskID uint64, linkType, meta string) { + f.task.MetaReady(ctx, taskID, linkType, meta) } diff --git a/framework/data_request.go b/framework/data_request.go index 85be0c8..a17eab0 100644 --- a/framework/data_request.go +++ b/framework/data_request.go @@ -2,25 +2,42 @@ package framework import ( "fmt" - "strings" + "strconv" "time" + "github.com/golang/protobuf/proto" "github.com/taskgraph/taskgraph/pkg/etcdutil" "golang.org/x/net/context" "google.golang.org/grpc" + "google.golang.org/grpc/metadata" ) var ( ErrEpochMismatch = fmt.Errorf("server epoch mismatch") ) -func (f *framework) CheckEpoch(epoch uint64) error { +func (f *framework) CheckGRPCContext(ctx context.Context) error { + md, ok := metadata.FromContext(ctx) + if !ok { + return fmt.Errorf("Can't get grpc.Metadata from context: %v", ctx) + } + epoch, err := strconv.ParseUint(md["epoch"], 10, 64) + if err != nil { + return err + } + // send it to framework central select and check epoch. resChan := make(chan bool, 1) - f.epochCheckChan <- &epochCheck{ + // The select loop might stop running but we still need to return error to user. + // We can't use the ctx here because it's the grpc context. + select { + case f.epochCheckChan <- &epochCheck{ epoch: epoch, resChan: resChan, + }: + case <-f.globalStop: + return fmt.Errorf("framework stopped") } - ok := <-resChan + ok = <-resChan if ok { return nil } else { @@ -28,46 +45,79 @@ func (f *framework) CheckEpoch(epoch uint64) error { } } +func (f *framework) DataRequest(ctx context.Context, toID uint64, method string, input proto.Message) { + epoch, ok := ctx.Value(epochKey).(uint64) + if !ok { + f.log.Fatalf("Can not find epochKey or cast is in DataRequest") + } + // assumption here: + // Event driven task will call this in a synchronous way so that + // the epoch won't change at the time task sending this request. + // Epoch may change, however, before the request is actually being sent. + select { + case f.dataReqtoSendChan <- &dataRequest{ + ctx: f.makeGRPCContext(ctx), + taskID: toID, + epoch: epoch, + input: input, + method: method, + }: + case <-ctx.Done(): + f.log.Printf("abort data request, to %d, epoch %d, method %s", toID, epoch, method) + } +} + +// encode metadata to context in grpc specific way +func (f *framework) makeGRPCContext(ctx context.Context) context.Context { + md := metadata.MD{ + "taskID": strconv.FormatUint(f.taskID, 10), + "epoch": strconv.FormatUint(f.epoch, 10), + } + return metadata.NewContext(ctx, md) +} + func (f *framework) sendRequest(dr *dataRequest) { addr, err := etcdutil.GetAddress(f.etcdClient, f.name, dr.taskID) if err != nil { - // TODO: We should handle network faults later by retrying - f.log.Panicf("getAddress(%d) failed: %v", dr.taskID, err) + f.log.Printf("getAddress(%d) failed: %v", dr.taskID, err) + go f.retrySendRequest(dr) return } - - if dr.retry { - f.log.Printf("retry request data from task %d, addr %s", dr.taskID, addr) - } else { - f.log.Printf("request data from task %d, addr %s", dr.taskID, addr) - } - + // TODO: reuse ClientConn and reply message. + // The grpc.WithTimeout would help detect any disconnection in failfast. + // Otherwise grpc.Invoke will keep retrying. cc, err := grpc.Dial(addr, grpc.WithTimeout(heartbeatInterval)) // we need to retry if some task failed and there is a temporary Get request failure. if err != nil { - f.log.Printf("grpc.Dial from task %d (addr: %s) failed: %v", dr.taskID, addr, err) + f.log.Printf("grpc.Dial to task %d (addr: %s) failed: %v", dr.taskID, addr, err) // Should retry for other errors. go f.retrySendRequest(dr) return } defer cc.Close() + if dr.retry { + f.log.Printf("retry data request %s to task %d, addr %s", dr.method, dr.taskID, addr) + } else { + f.log.Printf("data request %s to task %d, addr %s", dr.method, dr.taskID, addr) + } reply := f.task.CreateOutputMessage(dr.method) err = grpc.Invoke(dr.ctx, dr.method, dr.input, reply, cc) if err != nil { - if strings.Contains(err.Error(), "server epoch mismatch") { - // It's out of date. Should abort this data request. - return - } - f.log.Printf("grpc.Invoke from task %d (addr: %s), method: %s, failed: %v", dr.taskID, addr, dr.method, err) + f.log.Printf("grpc.Invoke to task %d (addr: %s), method: %s, failed: %v", dr.taskID, addr, dr.method, err) go f.retrySendRequest(dr) return } - f.dataRespChan <- &dataResponse{ + + select { + case f.dataRespChan <- &dataResponse{ epoch: dr.epoch, taskID: dr.taskID, method: dr.method, input: dr.input, output: reply, + }: + case <-dr.ctx.Done(): + f.log.Printf("abort data response, to %d, epoch %d, method %s", dr.taskID, dr.epoch, dr.method) } } @@ -76,7 +126,11 @@ func (f *framework) retrySendRequest(dr *dataRequest) { // gets up and running. time.Sleep(2 * heartbeatInterval) dr.retry = true - f.dataReqtoSendChan <- dr + select { + case f.dataReqtoSendChan <- dr: + case <-dr.ctx.Done(): + f.log.Printf("abort data request, to %d, epoch %d, method %s", dr.taskID, dr.epoch, dr.method) + } } // Framework http server for data request. @@ -85,9 +139,11 @@ func (f *framework) retrySendRequest(dr *dataRequest) { // On success, it should respond with requested data in http body. func (f *framework) startHTTP() { f.log.Printf("serving grpc on %s\n", f.ln.Addr()) - err := f.task.CreateServer().Serve(f.ln) + server := f.task.CreateServer() + err := server.Serve(f.ln) select { - case <-f.httpStop: + case <-f.globalStop: + server.Stop() f.log.Printf("grpc stops serving") default: if err != nil { @@ -96,13 +152,6 @@ func (f *framework) startHTTP() { } } -// Close listener, stop HTTP server; -// Write error message back to under-serving responses. -func (f *framework) stopHTTP() { - close(f.httpStop) - f.ln.Close() -} - func (f *framework) handleDataResp(ctx context.Context, resp *dataResponse) { f.task.DataReady(ctx, resp.taskID, resp.method, resp.output) } diff --git a/framework/event_type.go b/framework/event_type.go index 0a35fcc..6e042e9 100644 --- a/framework/event_type.go +++ b/framework/event_type.go @@ -6,10 +6,10 @@ import ( ) type metaChange struct { - from uint64 - who string - epoch uint64 - meta string + from uint64 + epoch uint64 + linkType string + meta string } type dataRequest struct { diff --git a/framework/framework.go b/framework/framework.go index 9690f48..10020b2 100644 --- a/framework/framework.go +++ b/framework/framework.go @@ -7,7 +7,6 @@ import ( "net" "github.com/coreos/go-etcd/etcd" - "github.com/golang/protobuf/proto" "github.com/taskgraph/taskgraph" "github.com/taskgraph/taskgraph/pkg/etcdutil" "golang.org/x/net/context" @@ -23,30 +22,33 @@ type framework struct { // user defined interfaces taskBuilder taskgraph.TaskBuilder - topology taskgraph.Topology - - task taskgraph.Task - taskID uint64 - epoch uint64 - etcdClient *etcd.Client - ln net.Listener + // user topology defines as a set of linkType topology, + // the key of map respresents the link type in our topology, like "Master", "Parents", + // for every key, the map contianer store the linkType topology as interface decleared + // and serves as decribing the linking by returning the corresponding taskID set + topology map[string]taskgraph.Topology + + task taskgraph.Task + taskID uint64 + epoch uint64 + etcdClient *etcd.Client + ln net.Listener + userCtx context.Context + userCtxCancel context.CancelFunc // A meta is a signal for specific epoch some task has some data. // However, our fault tolerance mechanism will start another task if it failed - // and flag the same meta again. Therefore, we keep track of notified meta. + // and flag the same meta again. Therefore, we keep track of notified meta. metaNotified map[string]bool // etcd stops - metaStops []chan bool - epochStop chan bool - - httpStop chan struct{} - heartbeatStop chan struct{} + metaWatchStop chan bool + epochWatchStop chan bool - epochPassed chan struct{} + globalStop chan struct{} // event loop - epochChan chan uint64 + epochWatcher chan uint64 metaChan chan *metaChange dataReqtoSendChan chan *dataRequest dataRespChan chan *dataResponse @@ -62,21 +64,24 @@ type contextKey int // different integer values. const epochKey contextKey = 1 -// Now use google context, for we simply create a barebone and attach the epoch to it. -func (f *framework) createContext() context.Context { - return context.WithValue(context.Background(), epochKey, f.epoch) -} - func (f *framework) FlagMeta(ctx context.Context, linkType, meta string) { epoch, ok := ctx.Value(epochKey).(uint64) if !ok { - f.log.Fatalf("Can not find epochKey in FlagMeta: %d", epoch) + f.log.Panicf("Can not find epochKey in FlagMeta, epoch: %d", epoch) } - value := fmt.Sprintf("%d-%s", epoch, meta) - _, err := f.etcdClient.Set(etcdutil.MetaPath(linkType, f.name, f.GetTaskID()), value, 0) - if err != nil { - f.log.Fatalf("etcdClient.Set failed; key: %s, value: %s, error: %v", - etcdutil.MetaPath(linkType, f.name, f.GetTaskID()), value, err) + // send the meta change notification to every task of specified link type. + for _, id := range f.topology[linkType].GetNeighbors(epoch) { + // The value is made of "epoch-fromID-metadata" + // + // Epoch is prepended to meta. When a new one starts and replaces + // the old one, it doesn't need to handle previous things, whose + // epoch is smaller than current one. + value := fmt.Sprintf("%d-%d-%s-%s", epoch, f.taskID, linkType, meta) + _, err := f.etcdClient.Set(etcdutil.MetaPath(f.name, id), value, 0) + if err != nil { + f.log.Panicf("etcdClient.Set failed; key: %s, value: %s, error: %v", + etcdutil.MetaPath(f.name, id), value, err) + } } } @@ -95,34 +100,12 @@ func (f *framework) IncEpoch(ctx context.Context) { } } -func (f *framework) DataRequest(ctx context.Context, toID uint64, method string, input proto.Message) { - epoch, ok := ctx.Value(epochKey).(uint64) - if !ok { - f.log.Fatalf("Can not find epochKey or cast is in DataRequest") - } - - // assumption here: - // Event driven task will call this in a synchronous way so that - // the epoch won't change at the time task sending this request. - // Epoch may change, however, before the request is actually being sent. - f.dataReqtoSendChan <- &dataRequest{ - ctx: ctx, - taskID: toID, - epoch: epoch, - input: input, - method: method, - } -} - -func (f *framework) GetTopology() taskgraph.Topology { return f.topology } - -// this will shutdown local node instead of global job. -func (f *framework) stop() { - close(f.epochChan) -} +func (f *framework) GetTopology() map[string]taskgraph.Topology { return f.topology } func (f *framework) Kill() { - f.stop() + // framework select loop will quit and end like getting a exit epoch, except that + // it won't set exit epoch across cluster. + close(f.epochWatcher) } // When node call this on framework, it simply set epoch to exitEpoch, diff --git a/framework/framework_test.go b/framework/framework_test.go index 9c10664..712065d 100644 --- a/framework/framework_test.go +++ b/framework/framework_test.go @@ -39,7 +39,8 @@ func TestRequestDataEpochMismatch(t *testing.T) { fw.SetTaskBuilder(&testableTaskBuilder{ setupLatch: &wg, }) - fw.SetTopology(topo.NewTreeTopology(1, 1)) + fw.AddLinkage("Parents", topo.NewTreeTopologyOfParent(1, 1)) + fw.AddLinkage("Children", topo.NewTreeTopologyOfChildren(1, 1)) wg.Add(1) go fw.Start() wg.Wait() @@ -62,8 +63,8 @@ func TestRequestDataEpochMismatch(t *testing.T) { // there will be some data transferring and captured by application task. // Here we have implemented a helper user task to capture those data, test if // it's passed from framework correctly and unmodified. -func TestFrameworkFlagMetaReady(t *testing.T) { - appName := "framework_test_flagmetaready" +func TestFrameworkFlagMeta(t *testing.T) { + appName := "TestFrameworkFlagMeta" etcdURLs := []string{"http://localhost:4001"} // launch controller to setup etcd layout ctl := controller.New(appName, etcd.NewClient(etcdURLs), 2, []string{"Parents", "Children"}) @@ -90,15 +91,16 @@ func TestFrameworkFlagMetaReady(t *testing.T) { var wg sync.WaitGroup taskBuilder := &testableTaskBuilder{ - dataMap: nil, cDataChan: cDataChan, pDataChan: pDataChan, setupLatch: &wg, } f0.SetTaskBuilder(taskBuilder) - f0.SetTopology(topo.NewTreeTopology(2, 2)) + f0.AddLinkage("Parents", topo.NewTreeTopologyOfParent(2, 2)) + f0.AddLinkage("Children", topo.NewTreeTopologyOfChildren(2, 2)) f1.SetTaskBuilder(taskBuilder) - f1.SetTopology(topo.NewTreeTopology(2, 2)) + f1.AddLinkage("Parents", topo.NewTreeTopologyOfParent(2, 2)) + f1.AddLinkage("Children", topo.NewTreeTopologyOfChildren(2, 2)) taskBuilder.setupLatch.Add(2) go f0.Start() @@ -121,19 +123,19 @@ func TestFrameworkFlagMetaReady(t *testing.T) { ctx := context.WithValue(context.Background(), epochKey, uint64(0)) for i, tt := range tests { // 0: F#FlagChildMetaReady -> 1: T#ParentMetaReady - f0.FlagMeta(ctx, "Parents", tt.cMeta) + f0.FlagMeta(ctx, "Children", tt.cMeta) // from child(1)'s view - data := <-pDataChan - expected := &tDataBundle{0, tt.cMeta, "", nil} + data := <-cDataChan + expected := &tDataBundle{id: 0, meta: tt.cMeta} if !reflect.DeepEqual(data, expected) { t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) } // 1: F#FlagParentMetaReady -> 0: T#ChildMetaReady - f1.FlagMeta(ctx, "Children", tt.pMeta) + f1.FlagMeta(ctx, "Parents", tt.pMeta) // from parent(0)'s view - data = <-cDataChan - expected = &tDataBundle{1, tt.pMeta, "", nil} + data = <-pDataChan + expected = &tDataBundle{id: 1, meta: tt.pMeta} if !reflect.DeepEqual(data, expected) { t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) } @@ -141,8 +143,7 @@ func TestFrameworkFlagMetaReady(t *testing.T) { } func TestFrameworkDataRequest(t *testing.T) { - t.Skip("TODO") - appName := "framework_test_flagmetaready" + appName := "framework_test_datarequest" etcdURLs := []string{"http://localhost:4001"} // launch controller to setup etcd layout ctl := controller.New(appName, etcd.NewClient(etcdURLs), 2, []string{"Parents", "Children"}) @@ -151,20 +152,6 @@ func TestFrameworkDataRequest(t *testing.T) { } defer ctl.DestroyEtcdLayout() - tests := []struct { - req string - resp []byte - }{ - {"request", []byte("response")}, - {"parameters", []byte{1, 2, 3}}, - {"gradient", []byte{4, 5, 6}}, - } - - dataMap := make(map[string][]byte) - for _, tt := range tests { - dataMap[tt.req] = tt.resp - } - pDataChan := make(chan *tDataBundle, 1) cDataChan := make(chan *tDataBundle, 1) // simulate two tasks on two nodes -- 0 and 1 @@ -182,15 +169,16 @@ func TestFrameworkDataRequest(t *testing.T) { var wg sync.WaitGroup taskBuilder := &testableTaskBuilder{ - dataMap: dataMap, cDataChan: cDataChan, pDataChan: pDataChan, setupLatch: &wg, } f0.SetTaskBuilder(taskBuilder) - f0.SetTopology(topo.NewTreeTopology(2, 2)) + f0.AddLinkage("Parents", topo.NewTreeTopologyOfParent(2, 2)) + f0.AddLinkage("Children", topo.NewTreeTopologyOfChildren(2, 2)) f1.SetTaskBuilder(taskBuilder) - f1.SetTopology(topo.NewTreeTopology(2, 2)) + f1.AddLinkage("Parents", topo.NewTreeTopologyOfParent(2, 2)) + f1.AddLinkage("Children", topo.NewTreeTopologyOfChildren(2, 2)) taskBuilder.setupLatch.Add(2) go f0.Start() @@ -202,51 +190,37 @@ func TestFrameworkDataRequest(t *testing.T) { defer f0.ShutdownJob() ctx := context.WithValue(context.Background(), epochKey, uint64(0)) - ctx = ctx - for i, tt := range tests { - tt = tt - // 0: F#DataRequest -> 1: T#ServeAsChild -> 0: T#ChildDataReady - // f0.DataRequest(ctx, 1, "Children", tt.req) - - // from child(1)'s view at 1: T#ServeAsChild - data := <-pDataChan - expected := &tDataBundle{0, "", data.req, nil} - if !reflect.DeepEqual(data, expected) { - t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) - } - // from parent(0)'s view at 0: T#ChildDataReady - data = <-cDataChan - expected = &tDataBundle{1, "", data.req, data.resp} - if !reflect.DeepEqual(data, expected) { - t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) - } - // 1: F#DataRequest -> 0: T#ServeAsParent -> 1: T#ParentDataReady - // f1.DataRequest(ctx, 0, "Parents", tt.req) - // from parent(0)'s view at 0: T#ServeAsParent - data = <-cDataChan - expected = &tDataBundle{1, "", data.req, nil} - if !reflect.DeepEqual(data, expected) { - t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) - } - // from child(1)'s view at 1: T#ParentDataReady - data = <-pDataChan - expected = &tDataBundle{0, "", data.req, data.resp} - if !reflect.DeepEqual(data, expected) { - t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) - } + f0.DataRequest(ctx, 1, "/proto.Regression/GetGradient", nil) + data := <-pDataChan + expected := &tDataBundle{ + id: 1, + method: "/proto.Regression/GetGradient", + output: &pb.Gradient{1}, + } + if !reflect.DeepEqual(data, expected) { + t.Errorf("data bundle want = %v, get = %v", expected, data) + } + f1.DataRequest(ctx, 0, "/proto.Regression/GetParameter", nil) + data = <-cDataChan + expected = &tDataBundle{ + id: 0, + method: "/proto.Regression/GetParameter", + output: &pb.Parameter{1}, + } + if !reflect.DeepEqual(data, expected) { + t.Errorf("data bundle want = %v, get = %v", expected, data) } } type tDataBundle struct { - id uint64 - meta string - req string - resp []byte + id uint64 + meta string + method string + output proto.Message } type testableTaskBuilder struct { - dataMap map[string][]byte cDataChan chan *tDataBundle pDataChan chan *tDataBundle setupLatch *sync.WaitGroup @@ -255,10 +229,10 @@ type testableTaskBuilder struct { func (b *testableTaskBuilder) GetTask(taskID uint64) taskgraph.Task { switch taskID { case 0: - return &testableTask{dataMap: b.dataMap, dataChan: b.cDataChan, + return &testableTask{dataChan: b.pDataChan, setupLatch: b.setupLatch} case 1: - return &testableTask{dataMap: b.dataMap, dataChan: b.pDataChan, + return &testableTask{dataChan: b.cDataChan, setupLatch: b.setupLatch} default: panic("unimplemented") @@ -269,8 +243,6 @@ type testableTask struct { id uint64 framework taskgraph.Framework setupLatch *sync.WaitGroup - // dataMap will be used to serve data according to request - dataMap map[string][]byte // This channel is used to convey data passed from framework back to the main // thread, for checking. Thus it's initialized and passed in from outside. @@ -287,26 +259,26 @@ func (t *testableTask) Init(taskID uint64, framework taskgraph.Framework) { t.setupLatch.Done() } } -func (t *testableTask) Exit() {} -func (t *testableTask) SetEpoch(ctx context.Context, epoch uint64) {} +func (t *testableTask) Exit() {} +func (t *testableTask) EnterEpoch(ctx context.Context, epoch uint64) {} func (t *testableTask) MetaReady(ctx context.Context, fromID uint64, linkType, meta string) { if t.dataChan != nil { - t.dataChan <- &tDataBundle{fromID, meta, "", nil} + t.dataChan <- &tDataBundle{id: fromID, meta: meta} } } func (t *testableTask) DataReady(ctx context.Context, fromID uint64, method string, output proto.Message) { - panic("") + t.dataChan <- &tDataBundle{id: fromID, method: method, output: output} } // These are payload rpc for application purpose. func (t *testableTask) GetParameter(ctx context.Context, input *pb.Input) (*pb.Parameter, error) { - panic("") + return &pb.Parameter{1}, nil } func (t *testableTask) GetGradient(ctx context.Context, input *pb.Input) (*pb.Gradient, error) { - panic("") + return &pb.Gradient{1}, nil } func (t *testableTask) CreateOutputMessage(methodName string) proto.Message { @@ -330,7 +302,7 @@ func (t *testableTask) CreateServer() *grpc.Server { func createListener(t *testing.T) net.Listener { l, err := net.Listen("tcp4", "127.0.0.1:0") if err != nil { - t.Fatalf("net.Listen(\"tcp4\", \"\") failed: %v", err) + t.Fatalf("create listener failed: %v", err) } return l } diff --git a/framework/healthy.go b/framework/healthy.go index b269496..edf7c2e 100644 --- a/framework/healthy.go +++ b/framework/healthy.go @@ -11,9 +11,9 @@ var ( ) func (f *framework) heartbeat() { - f.heartbeatStop = make(chan struct{}) + f.globalStop = make(chan struct{}) go func() { - err := etcdutil.Heartbeat(f.etcdClient, f.name, f.taskID, heartbeatInterval, f.heartbeatStop) + err := etcdutil.Heartbeat(f.etcdClient, f.name, f.taskID, heartbeatInterval, f.globalStop) if err != nil { f.log.Printf("Heartbeat stops with error: %v\n", err) } diff --git a/framework/master.go b/framework/master.go new file mode 100644 index 0000000..c5a506e --- /dev/null +++ b/framework/master.go @@ -0,0 +1,31 @@ +package framework + +import ( + "log" + "net" + + "github.com/coreos/go-etcd/etcd" + "github.com/taskgraph/taskgraph" +) + +type master struct { + job string + etcdURL []string + listener net.Listener + logger *log.Logger + task taskgraph.MasterTask + workerNum uint64 + etcdClient *etcd.Client + stopChan chan struct{} +} + +func NewMasterBoot(job string, etcdURL []string, ln net.Listener, logger *log.Logger, task taskgraph.MasterTask, workerNum uint64) taskgraph.Bootup { + return &master{ + job: job, + etcdURL: etcdURL, + listener: ln, + logger: logger, + task: task, + workerNum: workerNum, + } +} diff --git a/framework/master_boot.go b/framework/master_boot.go new file mode 100644 index 0000000..30648ba --- /dev/null +++ b/framework/master_boot.go @@ -0,0 +1,59 @@ +package framework + +import ( + "github.com/coreos/go-etcd/etcd" + "github.com/taskgraph/taskgraph/pkg/etcdutil" + "golang.org/x/net/context" +) + +// Master is guaranteed the first to run. It mainly does the following: +// - initialize internal structures. +// - set up etcd. Register master address. Set up layout. +// - start server. This is the user defined grpc server. +// - start event handling. Select and handle events, e.g. sending and receiving messages, +// finishing job, etc. +// - run user task. + +func (m *master) Start() { + m.init() + m.setupEtcd() + m.startServer() + go m.startEventHandling() + m.runUserTask() + m.stop() +} + +func (m *master) init() { + m.etcdClient = etcd.NewClient(m.etcdURL) + m.stopChan = make(chan struct{}) +} + +func (m *master) setupEtcd() { + // init layout + // register master's addr + m.etcdClient.Set(etcdutil.MasterPath(m.job), m.listener.Addr().String(), 0) +} + +func (m *master) startServer() { + s := m.task.CreateServer() + go s.Serve(m.listener) +} + +func (m *master) startEventHandling() { + for { + select { + case <-m.stopChan: + return + } + } +} + +func (m *master) runUserTask() { + m.task.Setup(m) + m.task.Run(context.Background()) +} + +func (m *master) stop() { + m.listener.Close() + close(m.stopChan) +} diff --git a/framework/master_boot_test.go b/framework/master_boot_test.go new file mode 100644 index 0000000..37e3cd7 --- /dev/null +++ b/framework/master_boot_test.go @@ -0,0 +1,28 @@ +package framework + +import ( + "testing" + + "github.com/coreos/go-etcd/etcd" + "github.com/taskgraph/taskgraph/pkg/etcdutil" +) + +// Master should register its address in etcd so that workers can find him. +func TestMasterSetupEtcd(t *testing.T) { + job := "TestMasterSetupEtcd" + etcdClient := etcd.NewClient([]string{"http://localhost:4001"}) + m := &master{ + job: job, + etcdClient: etcdClient, + listener: createListener(t), + } + m.setupEtcd() + resp, err := etcdClient.Get(etcdutil.MasterPath(job), false, false) + if err != nil { + t.Fatalf("etcdClient.Get failed: %v", err) + } + addr := m.listener.Addr().String() + if resp.Node.Value != addr { + t.Fatalf("Wrong address on etcd: want = %s, get = %s", addr, resp.Node.Value) + } +} diff --git a/framework/master_frame.go b/framework/master_frame.go new file mode 100644 index 0000000..c4182d6 --- /dev/null +++ b/framework/master_frame.go @@ -0,0 +1,14 @@ +package framework + +import ( + "github.com/golang/protobuf/proto" + "golang.org/x/net/context" +) + +func (m *master) NotifyWorker(ctx context.Context, workerID uint64, method string, input proto.Message) (proto.Message, error) { + panic("") +} + +func (m *master) Intercept(ctx context.Context, method string, input proto.Message) (proto.Message, error) { + panic("") +} diff --git a/framework/worker.go b/framework/worker.go new file mode 100644 index 0000000..6994fc4 --- /dev/null +++ b/framework/worker.go @@ -0,0 +1,31 @@ +package framework + +import ( + "log" + "net" + + "github.com/coreos/go-etcd/etcd" + "github.com/taskgraph/taskgraph" +) + +type worker struct { + id uint64 + job string + etcdURL []string + listener net.Listener + logger *log.Logger + task taskgraph.WorkerTask + etcdClient *etcd.Client + stopChan chan struct{} +} + +func NewWorkerBoot(job string, etcdURL []string, ln net.Listener, logger *log.Logger, task taskgraph.WorkerTask, id uint64) taskgraph.Bootup { + return &worker{ + job: job, + etcdURL: etcdURL, + listener: ln, + logger: logger, + task: task, + id: id, + } +} diff --git a/framework/worker_boot.go b/framework/worker_boot.go new file mode 100644 index 0000000..bc24d8d --- /dev/null +++ b/framework/worker_boot.go @@ -0,0 +1,52 @@ +package framework + +import ( + "github.com/coreos/go-etcd/etcd" + "github.com/taskgraph/taskgraph/pkg/etcdutil" + "golang.org/x/net/context" +) + +// Worker has similar workflow to master. Although they have different impl. + +func (w *worker) Start() { + w.init() + w.setupEtcd() + w.startServer() + go w.startEventHandling() + w.runUserTask() + w.stop() +} + +func (w *worker) init() { + w.etcdClient = etcd.NewClient(w.etcdURL) + w.stopChan = make(chan struct{}) +} + +func (w *worker) setupEtcd() { + // register worker's addr + w.etcdClient.Set(etcdutil.WorkerPath(w.job, w.id), w.listener.Addr().String(), 0) +} + +func (w *worker) startServer() { + s := w.task.CreateServer() + go s.Serve(w.listener) +} + +func (w *worker) startEventHandling() { + for { + select { + case <-w.stopChan: + return + } + } +} + +func (w *worker) runUserTask() { + w.task.Setup(w, w.id) + w.task.Run(context.Background()) +} + +func (w *worker) stop() { + w.listener.Close() + close(w.stopChan) +} diff --git a/framework/worker_boot_test.go b/framework/worker_boot_test.go new file mode 100644 index 0000000..c89cdaa --- /dev/null +++ b/framework/worker_boot_test.go @@ -0,0 +1,30 @@ +package framework + +import ( + "testing" + + "github.com/coreos/go-etcd/etcd" + "github.com/taskgraph/taskgraph/pkg/etcdutil" +) + +// Worker should register its address in etcd so that others can find him. +func TestWorkerSetupEtcd(t *testing.T) { + job := "TestWorkerSetupEtcd" + id := uint64(1) + etcdClient := etcd.NewClient([]string{"http://localhost:4001"}) + w := &worker{ + job: job, + etcdClient: etcdClient, + listener: createListener(t), + id: id, + } + w.setupEtcd() + resp, err := etcdClient.Get(etcdutil.WorkerPath(job, id), false, false) + if err != nil { + t.Fatalf("etcdClient.Get failed: %v", err) + } + addr := w.listener.Addr().String() + if resp.Node.Value != addr { + t.Fatalf("Wrong address on etcd: want = %s, get = %s", addr, resp.Node.Value) + } +} diff --git a/framework/worker_frame.go b/framework/worker_frame.go new file mode 100644 index 0000000..4ad2bca --- /dev/null +++ b/framework/worker_frame.go @@ -0,0 +1,18 @@ +package framework + +import ( + "github.com/golang/protobuf/proto" + "golang.org/x/net/context" +) + +func (w *worker) NotifyMaster(ctx context.Context, method string, input proto.Message) (proto.Message, error) { + panic("") +} + +func (w *worker) DataRequest(ctx context.Context, workerID uint64, method string, input proto.Message) (proto.Message, error) { + panic("") +} + +func (w *worker) Intercept(ctx context.Context, method string, input proto.Message) (proto.Message, error) { + panic("") +} diff --git a/framework_interface.go b/framework_interface.go index 280893d..fd23bc2 100644 --- a/framework_interface.go +++ b/framework_interface.go @@ -14,7 +14,9 @@ type Bootstrap interface { SetTaskBuilder(taskBuilder TaskBuilder) // This allow the application to specify how tasks are connection at each epoch - SetTopology(topology Topology) + + // This allow user add their own link type of Topology into framework topology + AddLinkage(linkType string, topology Topology) // After all the configure is done, driver need to call start so that all // nodes will get into the event loop to run the application. @@ -25,7 +27,7 @@ type Bootstrap interface { // high level features. type Framework interface { // This allow the task implementation query its neighbors. - GetTopology() Topology + GetTopology() map[string]Topology // Kill the framework itself. // As epoch changes, some nodes isn't needed anymore @@ -51,7 +53,7 @@ type Framework interface { // Request data from task toID with specified linkType and meta. DataRequest(ctx context.Context, toID uint64, method string, input proto.Message) - CheckEpoch(epoch uint64) error + CheckGRPCContext(ctx context.Context) error } // Note that framework can decide how update can be done, and how to serve the updatelog. @@ -60,3 +62,48 @@ type BackedUpFramework interface { // of one primary and some backup copies. Update(taskID uint64, log UpdateLog) } + +type Bootup interface { + // Blocking call to run the task until it finishes. + Start() +} + +type GRPCHandlerInterceptor interface { + // Currently grpc doesn't support interceptor functionality. We need to rely on user + // to call this at handler implementation. + // The workflow would be + // C:Notify -> S:Intercept -> S:OnNotify + Intercept(ctx context.Context, method string, input proto.Message) (proto.Message, error) +} + +// Master-worker paradigm: +// There are usually one master (we can make it fault tolerance) and many workers. +// Master is responsible for making global decision and assign work to individual workers. +// Startup: +// 1. master should start first. +// 2. workers start with a unique worker ID, and ask/notify the master for assignment. +// +// Why master and what should be done on master? +// Only master can make global decisions. Master should store the states (initial, updated, +// completed, etc.) of each workers and make decisions when state changes. This is important +// when task restart happens and reset state to "initial". +// +// What about worker? +// Workers do the actual computation and data flow. +// The framework keeps track of master address(es) in etcd. + +type MasterFrame interface { + // User can use this interface to simplify sending the messages to worker. By keeping + // track of workers' states, user can make decisions on logical worker and communicate it + // using proto messages. + NotifyWorker(ctx context.Context, workerID uint64, method string, input proto.Message) (proto.Message, error) + GRPCHandlerInterceptor +} + +type WorkerFrame interface { + // It usually send states, etc. information to master in order to get further decision. + NotifyMaster(ctx context.Context, method string, input proto.Message) (proto.Message, error) + // Worker-worker data flow + DataRequest(ctx context.Context, workerID uint64, method string, input proto.Message) (proto.Message, error) + GRPCHandlerInterceptor +} diff --git a/integration/common.go b/integration/common.go new file mode 100644 index 0000000..a1cdf4e --- /dev/null +++ b/integration/common.go @@ -0,0 +1,42 @@ +package integration + +import ( + "net" + "testing" + + "github.com/taskgraph/taskgraph" + "github.com/taskgraph/taskgraph/example/topo" + "github.com/taskgraph/taskgraph/framework" +) + +func createListener(t *testing.T) net.Listener { + l, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatalf("net.Listen(\"tcp4\", \"\") failed: %v", err) + } + return l +} + +// This is used to show how to drive the network. +func driveWithTreeTopo(t *testing.T, jobName string, etcds []string, ntask uint64, taskBuilder taskgraph.TaskBuilder) { + drive( + t, + jobName, + etcds, + taskBuilder, + map[string]taskgraph.Topology{ + "Parents": topo.NewTreeTopologyOfParent(2, ntask), + "Children": topo.NewTreeTopologyOfChildren(2, ntask), + }, + ) +} + +func drive(t *testing.T, jobName string, etcds []string, taskBuilder taskgraph.TaskBuilder, topo map[string]taskgraph.Topology) { + + bootstrap := framework.NewBootStrap(jobName, etcds, createListener(t), nil) + bootstrap.SetTaskBuilder(taskBuilder) + for i, _ := range topo { + bootstrap.AddLinkage(i, topo[i]) + } + bootstrap.Start() +} diff --git a/integration/node_failure_test.go b/integration/node_failure_test.go index 239c77a..512bea4 100644 --- a/integration/node_failure_test.go +++ b/integration/node_failure_test.go @@ -34,13 +34,13 @@ func TestMasterSetEpochFailure(t *testing.T) { NumberOfIterations: numOfIterations, } for i := uint64(0); i < numOfTasks; i++ { - go drive(t, job, etcdURLs, numOfTasks, taskBuilder) + go driveWithTreeTopo(t, job, etcdURLs, numOfTasks, taskBuilder) } if <-taskBuilder.NodeProducer { taskBuilder.MasterConfig = nil log.Println("Starting a new node") // this time we start a new bootstrap whose task master doesn't fail. - go drive(t, job, etcdURLs, numOfTasks, taskBuilder) + go driveWithTreeTopo(t, job, etcdURLs, numOfTasks, taskBuilder) } wantData := []int32{0, 105, 210, 315, 420, 525, 630, 735, 840, 945, 1050} @@ -97,7 +97,7 @@ func testSlaveFailure(t *testing.T, job string, slaveConfig map[string]string) { go func() { for _ = range taskBuilder.NodeProducer { log.Println("Starting a new node") - go drive(t, job, etcdURLs, numOfTasks, taskBuilder) + go driveWithTreeTopo(t, job, etcdURLs, numOfTasks, taskBuilder) } }() for i := uint64(0); i < numOfTasks; i++ { diff --git a/integration/regression_framework_test.go b/integration/regression_framework_test.go index b3515e9..658975a 100644 --- a/integration/regression_framework_test.go +++ b/integration/regression_framework_test.go @@ -1,15 +1,11 @@ package integration import ( - "net" "testing" "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph" "github.com/taskgraph/taskgraph/controller" "github.com/taskgraph/taskgraph/example/regression" - "github.com/taskgraph/taskgraph/example/topo" - "github.com/taskgraph/taskgraph/framework" ) func TestRegressionFramework(t *testing.T) { @@ -29,7 +25,7 @@ func TestRegressionFramework(t *testing.T) { NumberOfIterations: numOfIterations, } for i := uint64(0); i < numOfTasks; i++ { - go drive(t, job, etcdURLs, numOfTasks, taskBuilder) + go driveWithTreeTopo(t, job, etcdURLs, numOfTasks, taskBuilder) } wantData := []int32{0, 105, 210, 315, 420, 525, 630, 735, 840, 945, 1050} @@ -46,19 +42,3 @@ func TestRegressionFramework(t *testing.T) { controller.WaitForJobDone() controller.Stop() } - -func createListener(t *testing.T) net.Listener { - l, err := net.Listen("tcp4", "127.0.0.1:0") - if err != nil { - t.Fatalf("net.Listen(\"tcp4\", \"\") failed: %v", err) - } - return l -} - -// This is used to show how to drive the network. -func drive(t *testing.T, jobName string, etcds []string, ntask uint64, taskBuilder taskgraph.TaskBuilder) { - bootstrap := framework.NewBootStrap(jobName, etcds, createListener(t), nil) - bootstrap.SetTaskBuilder(taskBuilder) - bootstrap.SetTopology(topo.NewTreeTopology(2, ntask)) - bootstrap.Start() -} diff --git a/op/func_interface.go b/op/func_interface.go index 153620f..08a3abb 100644 --- a/op/func_interface.go +++ b/op/func_interface.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op // For more information read: http://ewencp.org/blog/golang-iterators/ // We need an reasonably efficient way to enumerate all indexes. @@ -26,6 +26,9 @@ type Parameter interface { // This allow one to enumerate through all parameters IndexIterator() IndexIterator + + // Return raw data pointer + Data() []float32 } // This func is useful to fill the parameter with the same value @@ -94,5 +97,5 @@ type StopCriteria interface { // with one point in the parameter space, and end with an optimal // point. Return true if we find optimal point. type Minimizer interface { - Minimize(function Function, stopCriteria StopCriteria, param Parameter) bool + Minimize(function Function, stopCriteria StopCriteria, param Parameter) (float32, error) } diff --git a/op/projected_gradient.go b/op/projected_gradient.go index 0b302aa..99b5d9d 100644 --- a/op/projected_gradient.go +++ b/op/projected_gradient.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op // This defines how we do projected gradient. type ProjectedGradient struct { @@ -23,7 +23,7 @@ func NewProjectedGradient(projector *Projection, beta, sigma, alpha float32) *Pr // This implementation is based on "Projected Gradient Methods for Non-negative Matrix // Factorization" by Chih-Jen Lin. Particularly it is based on the discription of an // improved projected gradient method in page 10 of that paper. -func (pg *ProjectedGradient) Minimize(loss Function, stop StopCriteria, vec Parameter) bool { +func (pg *ProjectedGradient) Minimize(loss Function, stop StopCriteria, vec Parameter) (float32, error) { stt := vec @@ -31,43 +31,35 @@ func (pg *ProjectedGradient) Minimize(loss Function, stop StopCriteria, vec Para pg.projector.ClipPoint(stt) nxt := stt.CloneWithoutCopy() - cdd := stt.CloneWithoutCopy() Fill(nxt, 0) // Evaluate once ovalgrad := &vgpair{value: 0, gradient: stt.CloneWithoutCopy()} evaluate(loss, stt, ovalgrad) nvalgrad := &vgpair{value: 0, gradient: stt.CloneWithoutCopy()} - tvalgrad := &vgpair{value: 0, gradient: stt.CloneWithoutCopy()} - alpha := pg.alpha + pg.projector.ClipGradient(stt, ovalgrad.gradient) + alpha_moves := []int{0, 0, 0, 0, 0} + current_alpha := pg.alpha for k := 0; !stop.Done(stt, ovalgrad.value, ovalgrad.gradient); k += 1 { + idx := k % len(alpha_moves) + alpha := current_alpha + alpha_moves[idx] = 0; - pg.projector.ClipGradient(stt, ovalgrad.gradient) newPoint(stt, nxt, ovalgrad.gradient, alpha, pg.projector) evaluate(loss, nxt, nvalgrad) if pg.isGoodStep(stt, nxt, ovalgrad, nvalgrad) { - newPoint(stt, cdd, ovalgrad.gradient, alpha/pg.beta, pg.projector) - evaluate(loss, cdd, tvalgrad) - for pg.isGoodStep(stt, cdd, ovalgrad, tvalgrad) { - { - tmp := nxt - nxt = cdd - cdd = tmp - } - { - tmp := nvalgrad - nvalgrad = tvalgrad - tvalgrad = tmp - } - // Now increase alpha as much as we can. + alpha_moves[idx] = 1 + move_sum := 0 + for i := 0; i < len(alpha_moves); i++ { + move_sum += alpha_moves[i]; + } + if (move_sum > 1) { alpha /= pg.beta - newPoint(stt, cdd, ovalgrad.gradient, alpha/pg.beta, pg.projector) - evaluate(loss, cdd, tvalgrad) - if pg.isTooClose(cdd, nxt) { - break + for i := 0; i < len(alpha_moves); i++ { + alpha_moves[i] = 0 } } } else { @@ -75,31 +67,31 @@ func (pg *ProjectedGradient) Minimize(loss Function, stop StopCriteria, vec Para // of the objective value. for !pg.isGoodStep(stt, nxt, ovalgrad, nvalgrad) { alpha *= pg.beta + alpha_moves[idx] -= 1 newPoint(stt, nxt, ovalgrad.gradient, alpha, pg.projector) evaluate(loss, nxt, nvalgrad) } } + current_alpha = alpha // Now we arrive at a point satisfies sufficient decrease condition. // Swap the wts and gradient for the next round. - { - tmp := stt - stt = nxt - nxt = tmp - } - { - tmp := ovalgrad - ovalgrad = nvalgrad - nvalgrad = tmp - } + stt, nxt = nxt, stt + ovalgrad, nvalgrad = nvalgrad, ovalgrad + pg.projector.ClipGradient(stt, ovalgrad.gradient) + } + + // Originally stt == vec, but stt may be swapped to newly created param (nxt). So copy it to output param here. + for it := vec.IndexIterator(); it.Next(); { + i := it.Index() + vec.Set(i, stt.Get(i)) } // This is so that we can reuse the step size in next round. - // XXX(baigang): pg.alpha will be preserved to solve a different PG minimization, i.e the counterpart in the alternating setup. Shall we keep this? - pg.alpha = alpha + // pg.alpha = current_alpha - // Simply return true to indicate the minimization is done. - return true + // Return the final loss func value and error. + return ovalgrad.value, nil } // This implements the sufficient decrease condition described in Eq (13) @@ -113,16 +105,6 @@ func (pg *ProjectedGradient) isGoodStep(owts, nwts Parameter, ovg, nvg *vgpair) return valdiff <= pg.sigma*float32(sum) } -func (pg *ProjectedGradient) isTooClose(owts, nwts Parameter) bool { - sum := float64(0) - for it := owts.IndexIterator(); it.Next(); { - i := it.Index() - diff := float64(owts.Get(i) - nwts.Get(i)) - sum += diff * diff - } - return sum < 1e-16*float64(owts.IndexIterator().Size()) -} - // This creates a new point based on current point, step size and gradient. func newPoint(owts, nwts, grad Parameter, alpha float32, projector *Projection) { for it := owts.IndexIterator(); it.Next(); { diff --git a/op/projected_gradient_test.go b/op/projected_gradient_test.go index 5d5f948..627fc70 100644 --- a/op/projected_gradient_test.go +++ b/op/projected_gradient_test.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op import ( "fmt" @@ -27,7 +27,7 @@ func TestPGMinimize0(t *testing.T) { proj := &Projection{lower_bound: lower, upper_bound: upper} minimizer := &ProjectedGradient{projector: proj, beta: 0.1, sigma: 0.01, alpha: 1.0} loss := &Rosenbrock{numOfCopies: copy, count: 0} - stpCount := MakeFixCountStopCriteria(1200) + stpCount := MakeFixCountStopCriteria(1500) stt0 := createParam(3.0, -2.0, copy) diff --git a/op/projection.go b/op/projection.go index ece0350..b61e32a 100644 --- a/op/projection.go +++ b/op/projection.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op // This implementaton is useful for type oneParameter struct { @@ -18,6 +18,11 @@ func (s *oneParameter) Add(index int, value float32) { panic("can not add value") } +func (s *oneParameter) Data() []float32 { + panic("can not get data pointer") + return nil +} + // This allow us to generate parameter with same width. func (s *oneParameter) CloneWithoutCopy() Parameter { return &oneParameter{value: s.value, size: s.size} diff --git a/op/rosenbrock.go b/op/rosenbrock.go index c54a27b..d88300b 100644 --- a/op/rosenbrock.go +++ b/op/rosenbrock.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op // Rosenbrock is used as standard test function for optimization. // f(x, y) = (1-x)^2 + 100(y-x^2)^2, it has a global minimum of 0 diff --git a/op/stop_criteria.go b/op/stop_criteria.go index 67108c4..f3f5006 100644 --- a/op/stop_criteria.go +++ b/op/stop_criteria.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op import ( "time" diff --git a/op/vec_parameter.go b/op/vec_parameter.go index 37093cf..74b8159 100644 --- a/op/vec_parameter.go +++ b/op/vec_parameter.go @@ -1,4 +1,4 @@ -package taskgraph_op +package op // We need to implement a slice based parameter @@ -57,3 +57,11 @@ func (s *sliceParameter) IndexIterator() IndexIterator { func NewVecParameter(size int) Parameter { return &sliceParameter{param: make([]float32, size, size)} } + +func NewVecParameterWithData(data []float32) Parameter { + return &sliceParameter{param: data} +} + +func (s *sliceParameter) Data() []float32 { + return s.param +} diff --git a/pkg/etcdutil/healthy.go b/pkg/etcdutil/healthy.go index 1b8ff69..31e0c87 100644 --- a/pkg/etcdutil/healthy.go +++ b/pkg/etcdutil/healthy.go @@ -1,7 +1,6 @@ package etcdutil import ( - "fmt" "log" "math/rand" "path" @@ -84,17 +83,22 @@ func WaitFreeTask(client *etcd.Client, name string, logger *log.Logger) (uint64, } }() var resp *etcd.Response - select { - case resp = <-respChan: - case <-time.After(10 * time.Second): - return 0, fmt.Errorf("WaitFailure timeout!") - } - idStr := path.Base(resp.Node.Key) - id, err := strconv.ParseUint(idStr, 10, 64) - if err != nil { - return 0, err + var waitTime uint64 = 0 + for { + select { + case resp = <-respChan: + idStr := path.Base(resp.Node.Key) + id, err := strconv.ParseUint(idStr, 10, 64) + if err != nil { + return 0, err + } + return id, nil + case <-time.After(10 * time.Second): + waitTime++ + logger.Printf("Node already wait failure for %d0s", waitTime) + } } - return id, nil + } func computeTTL(interval time.Duration) uint64 { diff --git a/pkg/etcdutil/layout.go b/pkg/etcdutil/layout.go index 752e073..fc68b7f 100644 --- a/pkg/etcdutil/layout.go +++ b/pkg/etcdutil/layout.go @@ -10,14 +10,16 @@ import ( // /{app}/epoch -> global value for epoch // /{app}/tasks/: register tasks under this directory // /{app}/tasks/{taskID}/{replicaID} -> pointer to nodes, 0 replicaID means master -// /{app}/tasks/{taskID}/parentMeta -// /{app}/tasks/{taskID}/childMeta +// /{app}/tasks/{taskID}/meta -> meta change notification // /{app}/healthy/{taskID} -> tasks' healthy condition // /{app}/nodes/: register nodes under this directory // /{app}/nodes/{nodeID}/address -> scheme://host:port/{path(if http)} // /{app}/nodes/{nodeID}/ttl -> keep alive timeout // /{app}/FreeTasks/{taskID} +// /{job}/master/{replicaID} +// /{job}/worker/{workerID} + const ( TasksDir = "tasks" NodesDir = "nodes" @@ -61,10 +63,17 @@ func TaskMasterPath(appName string, taskID uint64) string { return path.Join("/", appName, TasksDir, strconv.FormatUint(taskID, 10), TaskMaster) } -func MetaPath(linkType, appName string, taskID uint64) string { +func MetaPath(appName string, taskID uint64) string { return path.Join("/", appName, TasksDir, strconv.FormatUint(taskID, 10), - linkType) + "meta") +} + +func MasterPath(job string) string { + return path.Join("/", job, "master/0") +} +func WorkerPath(job string, id uint64) string { + return path.Join("/", job, strconv.FormatUint(id, 10)) } diff --git a/pkg/etcdutil/meta.go b/pkg/etcdutil/meta.go index 71d208b..3c52d42 100644 --- a/pkg/etcdutil/meta.go +++ b/pkg/etcdutil/meta.go @@ -2,20 +2,20 @@ package etcdutil import "github.com/coreos/go-etcd/etcd" -func WatchMeta(c *etcd.Client, taskID uint64, path string, stop chan bool, responseHandler func(*etcd.Response, uint64)) error { +func WatchMeta(c *etcd.Client, path string, stop chan bool, responseHandler func(*etcd.Response)) error { resp, err := c.Get(path, false, false) if err != nil { return err } // Get previous meta. We need to handle it. if resp.Node.Value != "" { - responseHandler(resp, taskID) + go responseHandler(resp) } receiver := make(chan *etcd.Response, 1) go c.Watch(path, resp.EtcdIndex+1, false, receiver, stop) go func(receiver chan *etcd.Response) { for resp := range receiver { - responseHandler(resp, taskID) + responseHandler(resp) } }(receiver) return nil diff --git a/pkg/etcdutil/task.go b/pkg/etcdutil/task.go index a327f2d..82ab66d 100644 --- a/pkg/etcdutil/task.go +++ b/pkg/etcdutil/task.go @@ -1,24 +1,27 @@ package etcdutil import ( - "log" "strconv" + "strings" "github.com/coreos/go-etcd/etcd" ) -func TryOccupyTask(client *etcd.Client, name string, taskID uint64, connection string) bool { +func TryOccupyTask(client *etcd.Client, name string, taskID uint64, connection string) (bool, error) { _, err := client.Create(TaskHealthyPath(name, taskID), "health", 3) if err != nil { - return false + if strings.Contains(err.Error(), "Key already exists") { + return false, nil + } + return false, err } idStr := strconv.FormatUint(taskID, 10) client.Delete(FreeTaskPath(name, idStr), false) _, err = client.Set(TaskMasterPath(name, taskID), connection, 0) if err != nil { - log.Fatal(err) + return false, err } - return true + return true, nil } // getAddress will return the host:port address of the service taking care of diff --git a/task_interface.go b/task_interface.go index 20b01f3..6da1cb4 100644 --- a/task_interface.go +++ b/task_interface.go @@ -20,7 +20,7 @@ type Task interface { // Framework tells user task what current epoch is. // This give the task an opportunity to cleanup and regroup. - SetEpoch(ctx context.Context, epoch uint64) + EnterEpoch(ctx context.Context, epoch uint64) // The meta/data notifications obey exactly-once semantics. Note that the same // meta string will be notified only once even if you flag the meta more than once. @@ -49,3 +49,30 @@ type Backupable interface { // one update the state of copy. Update(log UpdateLog) } + +type GRPCHelper interface { + CreateOutputMessage(methodName string) proto.Message + CreateServer() *grpc.Server +} + +// Master task is assumed to be fault tolerant. + +type MasterTask interface { + Setup(framework MasterFrame) + Run(ctx context.Context) + + // Corresponds to NotifyMaster + OnNotify(ctx context.Context, workerID uint64, method string, input proto.Message) (proto.Message, error) + GRPCHelper +} + +type WorkerTask interface { + Setup(framework WorkerFrame, workerID uint64) + Run(ctx context.Context) + + // Corresponds to NotifyWorker + OnNotify(ctx context.Context, method string, input proto.Message) (proto.Message, error) + // Corresponds to DataRequest + ServeData(ctx context.Context, workerID uint64, method string, input proto.Message) (proto.Message, error) + GRPCHelper +} diff --git a/topology_interface.go b/topology_interface.go index 8416dee..61daa77 100644 --- a/topology_interface.go +++ b/topology_interface.go @@ -1,14 +1,22 @@ /* The topology is a data structure implemented by application, and used by framework implementation to setup/manage the event according to -topology defined here. The main usage is: +a set of topology defined here. + +Each Topology describe a specific linkType in overall topology. +After SetTaskID(), GetNeighbors() return the result of this relationship of this taskID +For instance, a topology describe a master link, it will instruct +the framework the master link of node in each epoch. + +The main usage is: a. At beginning of the framework initialization for each task, framework call the SetTaskID so that the singleton Topology knows which taskID it represents. -b. At beginning of each epoch, the framework implementation (on each task) - will call GetLinkTypes and GetNeighbors with given epoch, so that it knows - how to setup watcher for node failures. +b. At beginning of each epoch, the framework implementation call GetNeighbors + with given epoch, so that it knows how to set watcher for node failure + for specific relationship */ + package taskgraph // The Topology will be implemented by the application. @@ -19,9 +27,6 @@ type Topology interface { // we can get the local topology for each epoch later. SetTaskID(taskID uint64) - // This returns the type of links this topology supports - GetLinkTypes() []string - // This returns the neighbors of given link for this node at this epoch. - GetNeighbors(linkType string, epoch uint64) []uint64 + GetNeighbors(epoch uint64) []uint64 }