diff --git a/.gitignore b/.gitignore index 4b97723..5dc8b34 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ # Folders _obj _test +.tmp # Architecture specific extensions/prefixes *.[568vq] diff --git a/.script/bin/etcd b/.script/bin/etcd deleted file mode 100755 index 501f996..0000000 Binary files a/.script/bin/etcd and /dev/null differ diff --git a/.script/bin/etcd-v2.0.0-linux-amd64/etcd b/.script/bin/etcd-v2.0.0-linux-amd64/etcd deleted file mode 100755 index 963a32e..0000000 Binary files a/.script/bin/etcd-v2.0.0-linux-amd64/etcd and /dev/null differ diff --git a/.script/test b/.script/test deleted file mode 100755 index 94419a3..0000000 --- a/.script/test +++ /dev/null @@ -1,57 +0,0 @@ -#!/bin/bash -e - -ETCDTESTDIR=".tmp/etcd-testdir" - -if [ ! -d ".tmp" ]; then - mkdir .tmp -fi - -cleanup() { - kill $(jobs -p) 2>/dev/null - rm -r $ETCDTESTDIR -} -# cleanup on any kind of exit -trap "cleanup" SIGINT SIGTERM EXIT - -is_etcd_up_on_4001() { - if curl -fs "http://localhost:4001/v2/machines" 2>/dev/null; then - return 0 - fi - return 1 -} - -if is_etcd_up_on_4001 ; then - echo "existing etcd on localhost:4001" - exit 1 -fi - -if [ -d "$ETCDTESTDIR" ]; then - rm -r $ETCDTESTDIR -fi - -if [[ "$OSTYPE" == "darwin"* ]]; then - .script/bin/etcd --data-dir $ETCDTESTDIR > /dev/null 2>&1 & -else - .script/bin/etcd-v2.0.0-linux-amd64/etcd --data-dir $ETCDTESTDIR > /dev/null 2>&1 & -fi - -for i in $(seq 10); do - sleep 1 - if is_etcd_up_on_4001; then - break - fi -done - -if is_etcd_up_on_4001 ; then - echo "etcd is running on localhost:4001" -else - echo "etcd failed to run on localhost:4001" - exit 1 -fi - -# testing -go test -v ./controller -go test -v ./example/regression -go test -v ./framework -go test -v ./integration - diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index b15e97d..0000000 --- a/.travis.yml +++ /dev/null @@ -1,13 +0,0 @@ -language: go - -go: - - 1.4 - -install: - - go get github.com/coreos/go-etcd/etcd - -before_script: - - mkdir .tmp - -script: - - .script/test diff --git a/README.md b/README.md index e580af0..f959f62 100644 --- a/README.md +++ b/README.md @@ -3,20 +3,18 @@ TaskGraph [![Build Status](https://travis-ci.org/taskgraph/taskgraph.svg)](https://travis-ci.org/taskgraph/taskgraph) -TaskGraph is a framework for writing fault tolerent distributed applications. It assumes that application consists of a network of tasks, which are inter-connected based on certain topology (hence graph). TaskGraph assume for each task (logical unit of work), there are one primary node, and zero or more backup nodes. TaskGraph then help with two types of node failure: failure happens to nodes from different task, failure happens to the nodes from the same task. Framework monitors the task/node's health, and take care of restarting the failed tasks, and also pass on a standard set of events (parent/children fail/restart, primary/backup switch) to task implementation so that it can do application dependent recovery. +TaskGraph is a framework for writing fault tolerent distributed ML applications. It assumes that application consists of a network of tasks, which are inter-connected based on certain topology (hence graph). TaskGraph assume for each task (logical unit of work), there are one primary node, and zero or more backup nodes. TaskGraph then help with two types of node failure: failure happens to nodes from different task, failure happens to the nodes from the same task. Framework monitors the task/node's health, take care of restarting the failed tasks, and continue the job in a fault tolerant manner. +TaskGraph supports a channel-joint model. Usually a task would want some data from other tasks, compute based on the data, and provide the result to other tasks. This is done by composing some joints doing the work and the channels they get and send data. Once a joint gets all the data it needs, it calls user-implemented compute function. There is a special kind of joint which doesn't have any out channels, called sync point. Once a sync point has finished its work, an iteration is finished. Every joint cares about its local work. But in overall thinking, there is a network of joints/tasks getting data computed and flown across one by one. -TaskGrpah 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. +Many ML jobs are iterative. In TaskGraph, it's an epoch for one iteration. TaskGraph will call user-implemented function at the beginning of each epoch to compose a specification mentioned above. Fault tolerance, consistency are handled by framework. Once an iteration is done, TaskGraph will start a new iteration unless user shutdown the job. +An TaskGraph application usually has three layers: -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. +1. In driver (main function), applicaiton need to configure the TaskBuilder, which specifies what task needs to be run and what it does. Finally, user will start the job. -1. In driver (main function), applicaiton need to configure the task graph. This include setting up TaskBuilder, which specify what task need to run as each node. One also need to specify the network topology which specify who connect to whom at each iteration. Then FrameWork.Start is called so that every node will get into event loop. +2. TaskGraph framework handles fault tolerance within the framework. -2. TaskGraph framework handles fault tolerency within the framework. It uses etcd and/or kubernetes for this purpose. It should also handle the communication between logic task so that it can hide the communication between master and hot standby. +3. Application need to implement Task interface, to specify what they do for work for each epoch. -3. Application need to implement Task interface, to specify how they should react to parent/child dia/restart event to carry out the correct application logic. Note that application developer need to implement TaskBuilder/Topology that suit their need (implementaion of these three interface are wired together in the driver). - -For an example of driver and task implementation, check dummy_task.go. - -Note, for now, the completion of TaskGraph is not defined explicitly. Instead, each application will have their way of exit based on application dependent logic. As an example, the above application can stop if task 0 stops. We have the hooks in Framework so any node can potentially exit the entire TaskGraph. We also have hook in Task so that task implementation get a change to save the work. +For an example of driver and task implementation, check example/ directory. \ No newline at end of file diff --git a/controller/controller.go b/controller/controller.go deleted file mode 100644 index 92b2385..0000000 --- a/controller/controller.go +++ /dev/null @@ -1,107 +0,0 @@ -package controller - -import ( - "log" - "os" - "strconv" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph/pkg/etcdutil" -) - -// This is the controller of a job. -// A job needs controller to setup etcd data layout, request -// cluster containers, etc. to setup framework to run. -type Controller struct { - name string - etcdclient *etcd.Client - numOfTasks uint64 - failDetectStop chan bool - logger *log.Logger - jobStatusChan chan string -} - -func New(name string, etcd *etcd.Client, numOfTasks uint64) *Controller { - return &Controller{ - name: name, - etcdclient: etcd, - numOfTasks: numOfTasks, - logger: log.New(os.Stdout, "", log.Lshortfile|log.Ltime|log.Ldate), - } -} - -// A controller typical workflow: -// 1. controller sets up etcd layout before any task starts running. -// 2. Being ready, controller lets other tasks to run and reports any failure found. -func (c *Controller) Start() error { - if err := c.InitEtcdLayout(); err != nil { - return err - } - // Currently no previous changes will be watches before watch is setup. - // We assumes that ttl is usually a few seconds. watch is setup before that. - go c.startFailureDetection() - c.logger.Printf("Controller starting, name: %s, numberOfTask: %d\n", c.name, c.numOfTasks) - return nil -} - -func (c *Controller) WaitForJobDone() error { - <-c.jobStatusChan - return nil -} - -func (c *Controller) Stop() error { - c.DestroyEtcdLayout() - c.stopFailureDetection() - c.logger.Printf("Controller stoping...\n") - return nil -} - -func (c *Controller) InitEtcdLayout() error { - // Initilize the job epoch to 0 - etcdutil.MustCreate(c.etcdclient, c.logger, etcdutil.EpochPath(c.name), "0", 0) - c.setupWatchOnJobStatus() - // initiate etcd data layout for tasks - // currently it creates as many unassigned tasks as task masters. - 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) - key = etcdutil.ParentMetaPath(c.name, i) - etcdutil.MustCreate(c.etcdclient, c.logger, key, "", 0) - key = etcdutil.ChildMetaPath(c.name, i) - etcdutil.MustCreate(c.etcdclient, c.logger, key, "", 0) - } - return nil -} - -func (c *Controller) DestroyEtcdLayout() error { - _, err := c.etcdclient.Delete("/", true) - return err -} - -func (c *Controller) startFailureDetection() error { - c.failDetectStop = make(chan bool, 1) - err := etcdutil.DetectFailure(c.etcdclient, c.name, c.failDetectStop) - if err != nil { - // We currently didn't handle outside. So we do some logging at least. - c.logger.Printf("DetectFailure returns error: %v", err) - } - return err -} - -func (c *Controller) setupWatchOnJobStatus() { - c.jobStatusChan = make(chan string, 1) - key := etcdutil.JobStatusPath(c.name) - resp := etcdutil.MustCreate(c.etcdclient, c.logger, key, "", 0) - go func() { - resp, err := c.etcdclient.Watch(key, resp.EtcdIndex+1, false, nil, nil) - if err != nil { - c.logger.Panicf("Watch on job status (%v) failed: %v", key, err) - } - c.jobStatusChan <- resp.Node.Value - }() -} - -func (c *Controller) stopFailureDetection() error { - c.failDetectStop <- true - return nil -} diff --git a/controller/controller_test.go b/controller/controller_test.go deleted file mode 100644 index 37b31ab..0000000 --- a/controller/controller_test.go +++ /dev/null @@ -1,44 +0,0 @@ -package controller - -import ( - "strconv" - "testing" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph/pkg/etcdutil" -) - -// etcd needs to be initialized beforehand -func TestControllerInitEtcdLayout(t *testing.T) { - etcdClient := etcd.NewClient([]string{"http://localhost:4001"}) - - tests := []struct { - name string - numberOfTasks uint64 - }{ - {"test-1", 2}, - {"test-2", 4}, - } - - for i, tt := range tests { - c := New(tt.name, etcdClient, tt.numberOfTasks) - c.InitEtcdLayout() - - for taskID := uint64(0); taskID < tt.numberOfTasks; taskID++ { - key := etcdutil.FreeTaskPath(c.name, strconv.FormatUint(taskID, 10)) - if _, err := etcdClient.Get(key, false, false); err != nil { - t.Errorf("task %d: etcdClient.Get %v failed: %v", i, key, err) - } - key = etcdutil.ParentMetaPath(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.ChildMetaPath(c.name, taskID) - if _, err := etcdClient.Get(key, false, false); err != nil { - t.Errorf("task %d: etcdClient.Get %v failed: %v", i, key, err) - } - } - - c.DestroyEtcdLayout() - } -} diff --git a/datum_interface.go b/datum_interface.go deleted file mode 100644 index cc877f4..0000000 --- a/datum_interface.go +++ /dev/null @@ -1,25 +0,0 @@ -package taskgraph - -// Datum is the interface for basic loading and transformation. -type Datum interface{} - -// DatumIerator allow one to iterate through all the datum in the set. -type DatumIterator interface { - HasNext() bool - Next() Datum -} - -// This can be used to build a sequence of Datum from source. -type DatumIteratorBuilder interface { - Build(path string) DatumIterator -} - -// Transform Datum from one format to another. -type DatumTransformer interface { - Transform(old Datum) Datum -} - -// DatumStore host a set of Datum in the memory. -type DatumStore struct { - Cache []Datum -} 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/.gitignore b/example/regression/.gitignore deleted file mode 100644 index 4119bb5..0000000 --- a/example/regression/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -regression -result.txt diff --git a/example/regression/data.go b/example/regression/data.go new file mode 100644 index 0000000..b3ff332 --- /dev/null +++ b/example/regression/data.go @@ -0,0 +1,12 @@ +package regression + +type data struct { +} + +func (d *data) Marshal() ([]byte, error) { + panic("") +} + +func unmarshalData(b []byte) *data { + panic("") +} diff --git a/example/regression/demo/README.md b/example/regression/demo/README.md deleted file mode 100644 index 58440ed..0000000 --- a/example/regression/demo/README.md +++ /dev/null @@ -1,35 +0,0 @@ -Regression Framework -====== - -How to run ------- - -### Build -``` -go build -``` -This should generate a binary call `regression`. - -### Start etcd server - -Download an etcd binary from: -https://github.com/coreos/etcd/releases - -Unzip, start it by doing -``` -./etcd -``` - -The example assumes the default port 4001 - -### Run regression framework - -Modify `run_regression.sh` file with the correct `ETCDBIN` setting. Then run - -``` -./run_regression.sh -``` - -### Clean up -A binary: `regression`; -An output file: `result.txt`; diff --git a/example/regression/demo/main.go b/example/regression/demo/main.go deleted file mode 100644 index 0302726..0000000 --- a/example/regression/demo/main.go +++ /dev/null @@ -1,54 +0,0 @@ -package main - -import ( - "flag" - "log" - "net" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph/controller" - "github.com/taskgraph/taskgraph/example/regression" - "github.com/taskgraph/taskgraph/example/topo" - "github.com/taskgraph/taskgraph/framework" -) - -func main() { - programType := flag.String("type", "", "(c) controller or (t) task") - job := flag.String("job", "", "job name") - etcdURLs := []string{"http://localhost:4001"} - flag.Parse() - - if *job == "" { - log.Fatalf("Please specify a job name") - } - - ntask := uint64(2) - switch *programType { - case "c": - log.Printf("controller") - controller := controller.New(*job, etcd.NewClient(etcdURLs), ntask) - controller.Start() - controller.WaitForJobDone() - case "t": - log.Printf("task") - bootstrap := framework.NewBootStrap(*job, etcdURLs, createListener(), nil) - taskBuilder := ®ression.SimpleTaskBuilder{ - GDataChan: make(chan int32, 11), - NumberOfIterations: 10, - MasterConfig: map[string]string{"writefile": "result.txt"}, - } - bootstrap.SetTaskBuilder(taskBuilder) - bootstrap.SetTopology(topo.NewTreeTopology(2, ntask)) - bootstrap.Start() - default: - log.Fatal("Please choose a type: (c) controller, (t) task") - } -} - -func createListener() net.Listener { - l, err := net.Listen("tcp4", "127.0.0.1:0") - if err != nil { - log.Fatalf("net.Listen(\"tcp4\", \"\") failed: %v", err) - } - return l -} diff --git a/example/regression/demo/run_regression.sh b/example/regression/demo/run_regression.sh deleted file mode 100755 index 778c997..0000000 --- a/example/regression/demo/run_regression.sh +++ /dev/null @@ -1,25 +0,0 @@ -#!/bin/sh - -ETCDBIN=$GOPATH/etcd-v2.0.0-darwin-amd64 -RESULT_FILE="result.txt" - -# clear etcd -$ETCDBIN/etcdctl rm --recursive / -rm $RESULT_FILE - -./regression -job="regression test" -type=c & - -# need to wait for controller to setup -sleep 1 -./regression -job="regression test" -type=t & -./regression -job="regression test" -type=t & - -while [ ! -e $RESULT_FILE ]; do - sleep 1 -done - -echo "Result has been written to file 'result.txt'. Reading the file:" -echo "===== RESULT =====" -cat $RESULT_FILE - -trap "kill 0" SIGINT SIGTERM EXIT \ No newline at end of file diff --git a/example/regression/joint.go b/example/regression/joint.go new file mode 100644 index 0000000..bea506b --- /dev/null +++ b/example/regression/joint.go @@ -0,0 +1,40 @@ +package regression + +import "github.com/taskgraph/taskgraph" + +type parameterJoint struct { + parameter *data +} + +func (proc *parameterJoint) Compute(ins []taskgraph.InboundChannel, outs []taskgraph.OutboundChannel) { + if proc.parameter == nil { + proc.parameter = unmarshalData(ins[0].Get()) + } + for _, child := range outs { + child.Put(proc.parameter) + } +} + +type gradientJoint struct { + parameter *data +} + +func (proc *gradientJoint) Compute(ins []taskgraph.InboundChannel, outs []taskgraph.OutboundChannel) { + // master task have parameter already. slave task doesn't have, so he needs to + // retrieve from others. + if proc.parameter == nil { + proc.parameter = unmarshalData(ins[0].Get()) + } + proc.createLocalGradient(proc.parameter) + for _, in := range ins { + childG := unmarshalData(in.Get()) + proc.updateLocalGradient(childG) + } + outs[0].Put(proc.localGradient()) +} + +func (proc *gradientJoint) createLocalGradient(parameter *data) {} +func (proc *gradientJoint) updateLocalGradient(childG *data) {} +func (proc *gradientJoint) localGradient() *data { + panic("") +} diff --git a/example/regression/master_task.go b/example/regression/master_task.go new file mode 100644 index 0000000..e84e4c4 --- /dev/null +++ b/example/regression/master_task.go @@ -0,0 +1,44 @@ +package regression + +import ( + "github.com/taskgraph/taskgraph" + "github.com/taskgraph/taskgraph/job" +) + +type masterTask struct { + taskCommon + parameter *data +} + +func (tk *masterTask) SetEpoch(ctx taskgraph.Context, epoch uint64) { + if epoch == tk.totalIteration { + tk.framework.ShutdownJob(job.SuccessStatus) + return + } + tk.setupParameterJoint(ctx) + tk.setupGradientJoint(ctx) +} + +func (tk *masterTask) setupParameterJoint(ctx taskgraph.Context) { + // It's a source point because it doesn't have any inbound chan. + cp := ctx.CreateComposer() + cp.SetJoint(¶meterJoint{ + parameter: tk.parameter, + }) + + for _, child := range tk.treeTopo.GetChildren() { + cp.CreateOutboundChannel(child, "parameter") + } +} + +func (tk *masterTask) setupGradientJoint(ctx taskgraph.Context) { + // This is a sync point because it doesn't have any outbound chan. + cp := ctx.CreateComposer() + cp.SetJoint(&gradientJoint{ + parameter: tk.parameter, + }) + + for _, child := range tk.treeTopo.GetChildren() { + cp.CreateInboundChannel(child, "gradient") + } +} diff --git a/example/regression/regression.go b/example/regression/regression.go index 63df4b8..2efa621 100644 --- a/example/regression/regression.go +++ b/example/regression/regression.go @@ -1,345 +1,13 @@ package regression -import ( - "encoding/json" - "fmt" - "io/ioutil" - "log" - "math/rand" - "os" - "strconv" +import "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph/pkg/common" -) - -/* -The dummy task is designed for regresion test of taskgraph framework. -This works with tree topology. -The main idea behind the regression test is following: -There will be two kinds of dummyTasks: master and slaves. We will have one master -sits at the top with taskID = 0, and then rest 6 (2^n - 2) tasks forms a tree under -the master. There will be 10 epochs, from 1 to 10, at each epoch, we send out a -vector with all values equal to epochID, and each slave is supposedly return a vector -with all values equals epochID*taskID, the values are reduced back to master, and -master will print out the epochID and aggregated vector. After all 10 epoch, it kills -job. -*/ - -// dummyData is used to carry parameter and gradient; -type dummyData struct { - Value int32 -} - -// dummyMaster is prototype of parameter server, for now it does not -// 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. -type dummyMaster struct { - dataChan chan int32 - NodeProducer chan bool - framework taskgraph.Framework - epoch, taskID uint64 - logger *log.Logger - config map[string]string - numberOfIterations uint64 - - param, gradient *dummyData - fromChildren map[uint64]*dummyData -} - -// This is useful to bring the task up to speed from scratch or if it recovers. -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.logger = log.New(ioutil.Discard, "", log.Ldate|log.Ltime|log.Lshortfile) -} -func (t *dummyMaster) Exit() {} - -// Ideally, we should also have the following: -func (t *dummyMaster) ParentMetaReady(ctx taskgraph.Context, parentID uint64, meta string) {} -func (t *dummyMaster) ChildMetaReady(ctx taskgraph.Context, childID uint64, meta string) { - t.logger.Printf("master ChildMetaReady, task: %d, epoch: %d, child: %d\n", t.taskID, t.epoch, childID) - // Get data from child. When all the data is back, starts the next epoch. - ctx.DataRequest(childID, meta) -} - -// This give the task an opportunity to cleanup and regroup. -func (t *dummyMaster) SetEpoch(ctx taskgraph.Context, epoch uint64) { - t.logger.Printf("master SetEpoch, task: %d, epoch: %d\n", t.taskID, epoch) - if t.testablyFail("SetEpoch", strconv.FormatUint(epoch, 10)) { - return - } - - t.param = &dummyData{} - t.gradient = &dummyData{} - - t.epoch = epoch - t.param.Value = int32(t.epoch) - - // Make sure we have a clean slate. - t.fromChildren = make(map[uint64]*dummyData) - ctx.FlagMetaToChild("ParamReady") -} - -// These are payload rpc for application purpose. -func (t *dummyMaster) ServeAsParent(fromID uint64, req string, dataReceiver chan<- []byte) { - b, err := json.Marshal(t.param) - if err != nil { - t.logger.Fatalf("Master can't encode parameter: %v, error: %v\n", t.param, err) - } - dataReceiver <- b -} - -func (t *dummyMaster) ServeAsChild(fromID uint64, req string, dataReceiver chan<- []byte) {} - -func (t *dummyMaster) ParentDataReady(ctx taskgraph.Context, parentID uint64, req string, resp []byte) { -} -func (t *dummyMaster) ChildDataReady(ctx taskgraph.Context, childID uint64, req string, resp []byte) { - d := new(dummyData) - json.Unmarshal(resp, d) - if _, ok := t.fromChildren[childID]; ok { - return - } - t.fromChildren[childID] = d - - 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().GetChildren(t.epoch)) { - 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) - ctx.IncEpoch() - } - } -} - -func (t *dummyMaster) testablyFail(method string, args ...string) bool { - if t.config == nil { - return false - } - if t.config[method] != "fail" { - return false - } - if len(args) >= 1 && t.config["failepoch"] != "" { - // we need to care about fail at specific epoch - if t.config["failepoch"] != args[0] { - return false - } - } - if !probablyFail(t.config["faillevel"]) { - return false - } - t.logger.Printf("master task %d testably fail, method: %s\n", t.taskID, method) - t.framework.Kill() - t.NodeProducer <- true - return true -} - -// dummySlave is an prototype for data shard in machine learning applications. -// It mainly does to things, pass on parameters to its children, and collect -// gradient back then add them together before make it available to its parent. -type dummySlave struct { - framework taskgraph.Framework - epoch, taskID uint64 - logger *log.Logger - NodeProducer chan bool - config map[string]string - - param, gradient *dummyData - fromChildren map[uint64]*dummyData - gradientReady *common.CountdownLatch +type regressionTaskBuilder struct { } -// This is useful to bring the task up to speed from scratch or if it recovers. -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.logger = log.New(ioutil.Discard, "", log.Ldate|log.Ltime|log.Lshortfile) -} -func (t *dummySlave) Exit() {} - -// Ideally, we should also have the following: -func (t *dummySlave) ParentMetaReady(ctx taskgraph.Context, parentID uint64, meta string) { - t.logger.Printf("slave ParentMetaReady, task: %d, epoch: %d\n", t.taskID, t.epoch) - ctx.DataRequest(parentID, meta) -} - -func (t *dummySlave) ChildMetaReady(ctx taskgraph.Context, childID uint64, meta string) { - 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() - ctx.DataRequest(childID, meta) - }() -} - -// This give the task an opportunity to cleanup and regroup. -func (t *dummySlave) SetEpoch(ctx taskgraph.Context, epoch uint64) { - t.logger.Printf("slave SetEpoch, task: %d, epoch: %d\n", t.taskID, epoch) - t.param = &dummyData{} - t.gradient = &dummyData{} - t.gradientReady = common.NewCountdownLatch(1) - - t.epoch = epoch - // Make sure we have a clean slate. - t.fromChildren = make(map[uint64]*dummyData) -} - -// These are payload rpc for application purpose. -func (t *dummySlave) ServeAsParent(fromID uint64, req string, dataReceiver chan<- []byte) { - b, err := json.Marshal(t.param) - if err != nil { - t.logger.Fatalf("Slave can't encode parameter: %v, error: %v\n", t.param, err) - } - dataReceiver <- b -} - -func (t *dummySlave) ServeAsChild(fromID uint64, req string, dataReceiver chan<- []byte) { - b, err := json.Marshal(t.gradient) - if err != nil { - t.logger.Fatalf("Slave can't encode gradient: %v, error: %v\n", t.gradient, err) - } - dataReceiver <- b -} - -func (t *dummySlave) ParentDataReady(ctx taskgraph.Context, parentID uint64, req string, resp []byte) { - t.logger.Printf("slave ParentDataReady, task: %d, epoch: %d, parent: %d\n", t.taskID, t.epoch, parentID) - if t.testablyFail("ParentDataReady") { - return - } - if t.gradientReady.Count() == 0 { - return - } - t.param = new(dummyData) - json.Unmarshal(resp, t.param) - // 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().GetChildren(t.epoch) - if len(children) != 0 { - ctx.FlagMetaToChild("ParamReady") - } else { - // On leaf node, we can immediately return by and flag parent - // that this node is ready. - ctx.FlagMetaToParent("GradientReady") - } -} - -func (t *dummySlave) ChildDataReady(ctx taskgraph.Context, childID uint64, req string, resp []byte) { - d := new(dummyData) - json.Unmarshal(resp, d) - if _, ok := t.fromChildren[childID]; ok { - return - } - t.fromChildren[childID] = d - - 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().GetChildren(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 - } - - ctx.FlagMetaToParent("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 - } - } -} - -func (t *dummySlave) testablyFail(method string, args ...string) bool { - if t.config == nil { - return false - } - if t.config[method] != "fail" { - return false - } - if !probablyFail(t.config["faillevel"]) { - return false - } - t.logger.Printf("slave task %d testably fail, method: %s\n", t.taskID, method) - t.framework.Kill() - t.NodeProducer <- true - return true -} - -func probablyFail(levelStr string) bool { - level, err := strconv.Atoi(levelStr) - if err != nil { - return false - } - if level < rand.Intn(100)+1 { - return false - } - return true -} - -// used for testing -type SimpleTaskBuilder struct { - GDataChan chan int32 - NumberOfIterations uint64 - NodeProducer chan bool - MasterConfig map[string]string - SlaveConfig map[string]string -} - -// This method is called once by framework implementation to get the -// right task implementation for the node/task. It requires the taskID -// for current node, and also a global array of tasks. -func (tc SimpleTaskBuilder) GetTask(taskID uint64) taskgraph.Task { +func (tb *regressionTaskBuilder) Build(taskID uint64) taskgraph.Task { if taskID == 0 { - return &dummyMaster{ - dataChan: tc.GDataChan, - NodeProducer: tc.NodeProducer, - config: tc.MasterConfig, - numberOfIterations: tc.NumberOfIterations, - } - } - return &dummySlave{ - NodeProducer: tc.NodeProducer, - config: tc.SlaveConfig, + return &masterTask{} } + return &slaveTask{} } diff --git a/example/regression/slave_task.go b/example/regression/slave_task.go new file mode 100644 index 0000000..1033ce4 --- /dev/null +++ b/example/regression/slave_task.go @@ -0,0 +1,42 @@ +package regression + +import "github.com/taskgraph/taskgraph" + +type slaveTask struct { + taskCommon +} + +func (tk *slaveTask) SetEpoch(ctx taskgraph.Context, epoch uint64) { + if epoch == tk.totalIteration { + return + } + tk.setupParameterJoint(ctx) + tk.setupGradientJoint(ctx) +} + +func (tk *slaveTask) setupParameterJoint(ctx taskgraph.Context) { + cp := ctx.CreateComposer() + cp.SetJoint(¶meterJoint{}) + + for _, parent := range tk.treeTopo.GetParents() { + cp.CreateInboundChannel(parent, "parameter") + } + + for _, child := range tk.treeTopo.GetChildren() { + cp.CreateOutboundChannel(child, "parameter") + } +} + +func (tk *slaveTask) setupGradientJoint(ctx taskgraph.Context) { + cp := ctx.CreateComposer() + cp.SetJoint(&gradientJoint{}) + + for _, child := range tk.treeTopo.GetChildren() { + cp.CreateInboundChannel(child, "gradient") + } + + for _, parent := range tk.treeTopo.GetParents() { + cp.CreateInboundChannel(parent, "parameter") + cp.CreateOutboundChannel(parent, "gradient") + } +} diff --git a/example/regression/task_common.go b/example/regression/task_common.go new file mode 100644 index 0000000..92d7243 --- /dev/null +++ b/example/regression/task_common.go @@ -0,0 +1,27 @@ +package regression + +import ( + "log" + "os" + + "github.com/taskgraph/taskgraph" + "github.com/taskgraph/taskgraph/example/topo" +) + +type taskCommon struct { + taskID uint64 + logger *log.Logger + framework taskgraph.Framework + treeTopo *topo.TreeTopology + totalIteration uint64 +} + +func (tk *taskCommon) Init(framework taskgraph.Framework, numberOfTasks uint64) { + tk.taskID = framework.TaskID() + tk.framework = framework + tk.treeTopo = topo.NewTreeTopology(2, numberOfTasks) + tk.treeTopo.SetTaskID(numberOfTasks) + tk.logger = log.New(os.Stdout, "", log.Ldate|log.Ltime|log.Lshortfile) +} + +func (tk *taskCommon) Exit() {} diff --git a/example/topo/full_topo.go b/example/topo/full_topo.go deleted file mode 100644 index 9ab1de8..0000000 --- a/example/topo/full_topo.go +++ /dev/null @@ -1,49 +0,0 @@ -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. -// -//Also the star structure stays the same between epochs. -type FullTopology struct { - numOfTasks uint64 - taskID uint64 - parents, children []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.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) - } - } - -} - -func (t *FullTopology) GetParents(epoch uint64) []uint64 { return t.parents } - -func (t *FullTopology) GetChildren(epoch uint64) []uint64 { return t.children } - -// 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{ - numOfTasks: nTasks, - } - return m -} diff --git a/example/topo/tree_topo.go b/example/topo/tree_topo.go index 8838d26..ed32077 100644 --- a/example/topo/tree_topo.go +++ b/example/topo/tree_topo.go @@ -29,9 +29,9 @@ func (t *TreeTopology) SetTaskID(taskID uint64) { } } -func (t *TreeTopology) GetParents(epoch uint64) []uint64 { return t.parents } +func (t *TreeTopology) GetParents() []uint64 { return t.parents } -func (t *TreeTopology) GetChildren(epoch uint64) []uint64 { return t.children } +func (t *TreeTopology) GetChildren() []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. diff --git a/example/topo/tree_topo_test.go b/example/topo/tree_topo_test.go index 78fb855..c1c5704 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) - parents := treeTopology.GetParents(0) + parents := treeTopology.GetParents() if len(parents) != len(tt.parents) { t.Errorf("TreeTopology27 got wrong number of parents for %q", tt.id) } @@ -72,7 +72,7 @@ func testTreeTopology(fanout, number uint64, tests []treeTopoTest, t *testing.T) } } - children := treeTopology.GetChildren(0) + children := treeTopology.GetChildren() if len(children) != len(tt.children) { t.Errorf("TreeTopology27 got wrong number of children for %q", tt.id) } diff --git a/factory/boostrap.go b/factory/boostrap.go new file mode 100644 index 0000000..7a04034 --- /dev/null +++ b/factory/boostrap.go @@ -0,0 +1,7 @@ +package factory + +import "github.com/taskgraph/taskgraph" + +func CreateBootstrap() taskgraph.Bootstrap { + panic("") +} diff --git a/framework/bootstrap.go b/framework/bootstrap.go deleted file mode 100644 index 34a977a..0000000 --- a/framework/bootstrap.go +++ /dev/null @@ -1,268 +0,0 @@ -package framework - -import ( - "fmt" - "log" - "net" - "os" - "strconv" - "strings" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph/framework/frameworkhttp" - "github.com/taskgraph/taskgraph/pkg/etcdutil" -) - -type taskRole int - -const ( - roleNone taskRole = iota - roleParent - roleChild -) - -// One need to pass in at least these two for framework to start. -func NewBootStrap(jobName string, etcdURLs []string, ln net.Listener, logger *log.Logger) taskgraph.Bootstrap { - return &framework{ - name: jobName, - etcdURLs: etcdURLs, - ln: ln, - log: logger, - } -} - -func (f *framework) SetTaskBuilder(taskBuilder taskgraph.TaskBuilder) { f.taskBuilder = taskBuilder } - -func (f *framework) SetTopology(topology taskgraph.Topology) { f.topology = topology } - -func (f *framework) Start() { - var err error - - if f.log == nil { - f.log = log.New(os.Stdout, "", log.Lshortfile|log.Ltime|log.Ldate) - } - - f.etcdClient = etcd.NewClient(f.etcdURLs) - - if err = f.occupyTask(); err != nil { - f.log.Panicf("occupyTask() failed: %v", err) - } - - 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 - // 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) - 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 - return - } - f.log.Printf("starting at epoch %d\n", f.epoch) - - // task builder and topology are defined by applications. - // 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) - - go f.startHTTP() - - f.heartbeat() - f.setupChannels() - 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.dataReqChan = make(chan *dataRequest, 100) - f.dataRespToSendChan = make(chan *dataResponse, 100) - f.dataRespChan = make(chan *frameworkhttp.DataResponse, 100) -} - -func (f *framework) run() { - f.log.Printf("framework starts to run") - defer f.log.Printf("framework stops running.") - f.setEpochStarted() - // this for-select is primarily used to synchronize epoch specific events. - for { - select { - case nextEpoch, ok := <-f.epochChan: - f.releaseEpochResource() - if !ok { // single task exit - nextEpoch = exitEpoch - return - } - f.epoch = nextEpoch - if f.epoch == exitEpoch { - return - } - // start the next epoch's work - f.setEpochStarted() - case meta := <-f.metaChan: - if meta.epoch != f.epoch { - break - } - // We need to create a context before handling next event. The context saves - // 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.who, meta.from, 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) - break - } - go f.sendRequest(req) - case req := <-f.dataReqChan: - if req.epoch != f.epoch { - f.log.Printf("epoch mismatch: request epoch: %d, current epoch: %d", req.epoch, f.epoch) - req.notifyEpochMismatch() - break - } - f.handleDataReq(req) - case resp := <-f.dataRespToSendChan: - if resp.epoch != f.epoch { - f.log.Printf("epoch mismatch: resp-to-send epoch: %d, current epoch: %d", resp.epoch, f.epoch) - resp.notifyEpochMismatch() - break - } - go f.sendResponse(resp) - case resp := <-f.dataRespChan: - if resp.Epoch != f.epoch { - f.log.Printf("epoch mismatch: response epoch: %d, current epoch: %d", resp.Epoch, f.epoch) - break - } - f.handleDataResp(f.createContext(), resp) - } - } -} - -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) - // setup etcd watches - // - create self's parent and child meta flag - // - watch parents' child meta flag - // - watch children's parent meta flag - f.watchMeta(roleParent, f.topology.GetParents(f.epoch)) - f.watchMeta(roleChild, f.topology.GetChildren(f.epoch)) -} - -func (f *framework) releaseEpochResource() { - close(f.epochPassed) - for _, c := range f.metaStops { - c <- true - } - f.metaStops = nil -} - -// 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() -} - -// occupyTask will grab the first unassigned task and register itself on etcd. -func (f *framework) occupyTask() error { - for { - freeTask, err := etcdutil.WaitFreeTask(f.etcdClient, f.name, f.log) - if err != nil { - return err - } - f.log.Printf("standby grabbed free task %d", freeTask) - ok := etcdutil.TryOccupyTask(f.etcdClient, f.name, freeTask, f.ln.Addr().String()) - if ok { - f.taskID = freeTask - return nil - } - f.log.Printf("standby tried task %d failed. Wait free task again.", freeTask) - } -} - -func (f *framework) watchMeta(who taskRole, taskIDs []uint64) { - stops := make([]chan bool, len(taskIDs)) - - for i, taskID := range taskIDs { - stop := make(chan bool, 1) - stops[i] = stop - - var watchPath string - switch who { - case roleParent: - // Watch parent's child-meta. - watchPath = etcdutil.ChildMetaPath(f.name, taskID) - case roleChild: - // Watch child's parent-meta. - watchPath = etcdutil.ParentMetaPath(f.name, taskID) - default: - f.log.Panic("unexpected role") - } - - // 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: who, - epoch: ep, - meta: values[1], - } - } - - // Need to pass in taskID to make it work. Didn't know why. - err := etcdutil.WatchMeta(f.etcdClient, taskID, watchPath, stop, responseHandler) - if err != nil { - f.log.Panicf("WatchMeta failed. path: %s, err: %v", watchPath, err) - } - } - f.metaStops = append(f.metaStops, stops...) -} - -func (f *framework) handleMetaChange(ctx taskgraph.Context, who taskRole, taskID uint64, meta string) { - // check if meta is handled before. - tm := taskMeta(taskID, meta) - if _, ok := f.metaNotified[tm]; ok { - return - } - f.metaNotified[tm] = true - - switch who { - case roleParent: - f.task.ParentMetaReady(ctx, taskID, meta) - case roleChild: - f.task.ChildMetaReady(ctx, taskID, meta) - } -} - -func taskMeta(taskID uint64, meta string) string { - return fmt.Sprintf("%s-%s", strconv.FormatUint(taskID, 10), meta) -} diff --git a/framework/channel/client.go b/framework/channel/client.go new file mode 100644 index 0000000..c31f39e --- /dev/null +++ b/framework/channel/client.go @@ -0,0 +1,33 @@ +package channel + +import ( + pb "github.com/taskgraph/taskgraph/framework/channel/proto" + "golang.org/x/net/context" +) + +type client struct { + inboundMap map[string][]*inbound + grpcClient pb.ChannelClient +} + +func (c *client) SetupDispatch() { + for tag, inbounds := range c.inboundMap { + go func(tag string, inbounds []*inbound) { + data, err := c.grpcClient.GetData(context.Background(), &pb.Tag{tag}) + if err != nil { + panic("") + } + for _, inbound := range inbounds { + inbound.dataChan <- data.Payload + } + }(tag, inbounds) + } +} + +func (c *client) AttachInbound(in *inbound) { + c.inboundMap[in.ID()] = append(c.inboundMap[in.ID()], in) +} + +func NewClient() *client { + +} diff --git a/framework/channel/common.go b/framework/channel/common.go new file mode 100644 index 0000000..507c2f9 --- /dev/null +++ b/framework/channel/common.go @@ -0,0 +1,20 @@ +package channel + +import "fmt" + +type channelCommon struct { + taskID uint64 + tag string +} + +func (c channelCommon) TaskID() uint64 { + return c.taskID +} + +func (c channelCommon) Tag() string { + return c.tag +} + +func (c channelCommon) ID() string { + return fmt.Sprintf("%d-%s", c.taskID, c.tag) +} diff --git a/framework/channel/inbound.go b/framework/channel/inbound.go new file mode 100644 index 0000000..3764fda --- /dev/null +++ b/framework/channel/inbound.go @@ -0,0 +1,14 @@ +package channel + +type inbound struct { + channelCommon + dataChan chan []byte +} + +func (c *inbound) Get() []byte { + return <-c.dataChan +} + +func NewInbound() *inbound { + +} diff --git a/framework/channel/outbound.go b/framework/channel/outbound.go new file mode 100644 index 0000000..788e55f --- /dev/null +++ b/framework/channel/outbound.go @@ -0,0 +1,20 @@ +package channel + +import "github.com/taskgraph/taskgraph" + +type outbound struct { + channelCommon + dataChan chan []byte +} + +func (c *outbound) Put(m taskgraph.Marshaler) { + data, err := m.Marshal() + if err != nil { + panic(err) + } + c.dataChan <- data +} + +func NewOutbound() *outbound { + +} diff --git a/framework/channel/proto/channel.pb.go b/framework/channel/proto/channel.pb.go new file mode 100644 index 0000000..42e5614 --- /dev/null +++ b/framework/channel/proto/channel.pb.go @@ -0,0 +1,105 @@ +// Code generated by protoc-gen-go. +// source: channel.proto +// DO NOT EDIT! + +/* +Package proto is a generated protocol buffer package. + +It is generated from these files: + channel.proto + +It has these top-level messages: + Tag + Data +*/ +package proto + +import proto1 "github.com/golang/protobuf/proto" + +import ( + context "golang.org/x/net/context" + grpc "google.golang.org/grpc" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto1.Marshal + +type Tag struct { + Name string `protobuf:"bytes,1,opt,name=name" json:"name,omitempty"` +} + +func (m *Tag) Reset() { *m = Tag{} } +func (m *Tag) String() string { return proto1.CompactTextString(m) } +func (*Tag) ProtoMessage() {} + +type Data struct { + Payload []byte `protobuf:"bytes,1,opt,name=payload,proto3" json:"payload,omitempty"` +} + +func (m *Data) Reset() { *m = Data{} } +func (m *Data) String() string { return proto1.CompactTextString(m) } +func (*Data) ProtoMessage() {} + +func init() { +} + +// Client API for Channel service + +type ChannelClient interface { + GetData(ctx context.Context, in *Tag, opts ...grpc.CallOption) (*Data, error) +} + +type channelClient struct { + cc *grpc.ClientConn +} + +func NewChannelClient(cc *grpc.ClientConn) ChannelClient { + return &channelClient{cc} +} + +func (c *channelClient) GetData(ctx context.Context, in *Tag, opts ...grpc.CallOption) (*Data, error) { + out := new(Data) + err := grpc.Invoke(ctx, "/proto.Channel/GetData", in, out, c.cc, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// Server API for Channel service + +type ChannelServer interface { + GetData(context.Context, *Tag) (*Data, error) +} + +func RegisterChannelServer(s *grpc.Server, srv ChannelServer) { + s.RegisterService(&_Channel_serviceDesc, srv) +} + +func _Channel_GetData_Handler(srv interface{}, ctx context.Context, buf []byte) (proto1.Message, error) { + in := new(Tag) + if err := proto1.Unmarshal(buf, in); err != nil { + return nil, err + } + out, err := srv.(ChannelServer).GetData(ctx, in) + if err != nil { + return nil, err + } + return out, nil +} + +var _Channel_serviceDesc = grpc.ServiceDesc{ + ServiceName: "proto.Channel", + HandlerType: (*ChannelServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "GetData", + Handler: _Channel_GetData_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, +} diff --git a/framework/channel/proto/channel.proto b/framework/channel/proto/channel.proto new file mode 100644 index 0000000..560eebe --- /dev/null +++ b/framework/channel/proto/channel.proto @@ -0,0 +1,15 @@ +syntax = "proto3"; + +package proto; + +service Channel { + rpc GetData (Tag) returns (Data) {} +} + +message Tag { + string name = 1; +} + +message Data { + bytes payload = 1; +} diff --git a/framework/channel/server.go b/framework/channel/server.go new file mode 100644 index 0000000..dad33bc --- /dev/null +++ b/framework/channel/server.go @@ -0,0 +1,24 @@ +package channel + +import ( + pb "github.com/taskgraph/taskgraph/framework/channel/proto" + "golang.org/x/net/context" +) + +type server struct { + outboundMap map[string]*outbound +} + +func (s *server) GetData(ctx context.Context, tag *pb.Tag) (*pb.Data, error) { + outbound := s.outboundMap[tag.Name] + data := <-outbound.dataChan + return &pb.Data{data}, nil +} + +func (s *server) AttachOutbound(out *outbound) { + s.outboundMap[out.ID()] = out +} + +func NewServer() *server { + +} diff --git a/framework/channel/simul_rgrsn_test.go b/framework/channel/simul_rgrsn_test.go new file mode 100644 index 0000000..669649a --- /dev/null +++ b/framework/channel/simul_rgrsn_test.go @@ -0,0 +1,35 @@ +package channel + +import "testing" + +// This test simulates regression framework joints uses. + +// A task can send a slave some parameter, and the slave have two joints that read +// the same parameter. +func TestParameterFanOut(t *testing.T) { + c := NewClient() + s := NewServer() + out := NewOutbound() + s.AttachOutbound(out) + in1 := NewInbound() + c.AttachInbound(in1) + in2 := NewInbound() + c.AttachInbound(in2) + c.SetupDispatch() + + go out.Put(new(dummyData)) + + if string(in1.Get()) != "taskgraph rocks" { + t.Errorf("inbound 1 gets = %s, wants = %s", string(in1.Get()), "taskgraph rocks") + } + if string(in2.Get()) != "taskgraph rocks" { + t.Errorf("inbound 2 gets = %s, wants = %s", string(in2.Get()), "taskgraph rocks") + } +} + +type dummyData struct { +} + +func (*dummyData) Marshal() ([]byte, error) { + return []byte("taskgraph rocks"), nil +} diff --git a/framework/context.go b/framework/context.go deleted file mode 100644 index b46e9ca..0000000 --- a/framework/context.go +++ /dev/null @@ -1,29 +0,0 @@ -package framework - -type context struct { - epoch uint64 - f *framework -} - -func (f *framework) createContext() *context { - return &context{ - epoch: f.epoch, - f: f, - } -} - -func (c *context) FlagMetaToParent(meta string) { - c.f.flagMetaToParent(meta, c.epoch) -} - -func (c *context) FlagMetaToChild(meta string) { - c.f.flagMetaToChild(meta, c.epoch) -} - -func (c *context) IncEpoch() { - c.f.incEpoch(c.epoch) -} - -func (c *context) DataRequest(toID uint64, req string) { - c.f.dataRequest(toID, req, c.epoch) -} diff --git a/framework/data_request.go b/framework/data_request.go deleted file mode 100644 index bdcd577..0000000 --- a/framework/data_request.go +++ /dev/null @@ -1,149 +0,0 @@ -package framework - -import ( - "net/http" - "time" - - "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph/framework/frameworkhttp" - "github.com/taskgraph/taskgraph/pkg/etcdutil" - "github.com/taskgraph/taskgraph/pkg/topoutil" -) - -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.Fatalf("getAddress(%d) failed: %v", dr.taskID, err) - 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) - } - - d, err := frameworkhttp.RequestData(addr, dr.req, f.taskID, dr.taskID, dr.epoch, f.log) - // we need to retry if some task failed and there is a temporary Get request failure. - if err != nil { - f.log.Printf("RequestData from task %d (addr: %s) failed: %v", dr.taskID, addr, err) - if err == frameworkhttp.ErrReqEpochMismatch { - // It's out of date. Should wait for new epoch to set up. - return - } - // Should retry for other errors. - go func() { - // we try again after the previous task key expires and hopefully another task - // gets up and running. - time.Sleep(2 * heartbeatInterval) - dr.retry = true - f.dataReqtoSendChan <- dr - }() - return - } - f.dataRespChan <- d -} - -// This is used by the server side to handle data requests coming from remote. -func (f *framework) GetTaskData(taskID, epoch uint64, req string) ([]byte, error) { - dataChan := make(chan []byte, 1) - f.dataReqChan <- &dataRequest{ - taskID: taskID, - epoch: epoch, - req: req, - dataChan: dataChan, - } - - select { - case d, ok := <-dataChan: - if !ok { - // it assumes that only epoch mismatch will close the channel - return nil, frameworkhttp.ErrReqEpochMismatch - } - return d, nil - case <-f.httpStop: - // If a node stopped running and there is remaining requests, we need to - // respond error message back. It is used to let client routines stop blocking -- - // especially helpful in test cases. - - // This is used to drain the channel queue and get the rest notified. - <-f.dataReqChan - return nil, frameworkhttp.ErrServerClosed - } -} - -// Framework http server for data request. -// Each request will be in the format: "/datareq?taskID=XXX&req=XXX". -// "taskID" indicates the requesting task. "req" is the meta data for this request. -// On success, it should respond with requested data in http body. -func (f *framework) startHTTP() { - f.log.Printf("serving http on %s\n", f.ln.Addr()) - // TODO: http server graceful shutdown - handler := frameworkhttp.NewDataRequestHandler(f.log, f) - err := http.Serve(f.ln, handler) - select { - case <-f.httpStop: - f.log.Printf("http stops serving") - default: - if err != nil { - f.log.Fatalf("http.Serve() returns error: %v\n", err) - } - } -} - -// 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) sendResponse(dr *dataResponse) { - dr.dataChan <- dr.data -} - -func (f *framework) handleDataReq(dr *dataRequest) { - dataReceiver := make(chan []byte, 1) - switch { - case topoutil.IsParent(f.topology, dr.epoch, dr.taskID): - f.task.ServeAsChild(dr.taskID, dr.req, dataReceiver) - case topoutil.IsChild(f.topology, dr.epoch, dr.taskID): - f.task.ServeAsParent(dr.taskID, dr.req, dataReceiver) - default: - f.log.Panic("unexpected") - } - go func() { - select { - case data, ok := <-dataReceiver: - if !ok || data == nil { - return - } - // Getting the data from task could take a long time. We need to let - // the response-to-send go through event loop to check epoch. - f.dataRespToSendChan <- &dataResponse{ - taskID: dr.taskID, - epoch: dr.epoch, - req: dr.req, - data: data, - dataChan: dr.dataChan, - } - case <-f.epochPassed: - // We can't leave a go-routine to wait for the data channel forever. - // Users might forget to close the channel if they didn't want to do anything. - // We can clean it up in releaseEpochResource() as the epoch moves on. - // Because we won't be interested even though the data would come later. - } - }() -} - -func (f *framework) handleDataResp(ctx taskgraph.Context, resp *frameworkhttp.DataResponse) { - switch { - case topoutil.IsParent(f.topology, resp.Epoch, resp.TaskID): - f.task.ParentDataReady(ctx, resp.TaskID, resp.Req, resp.Data) - case topoutil.IsChild(f.topology, resp.Epoch, resp.TaskID): - f.task.ChildDataReady(ctx, resp.TaskID, resp.Req, resp.Data) - default: - f.log.Panic("unexpected") - } -} diff --git a/framework/event_type.go b/framework/event_type.go deleted file mode 100644 index a409d40..0000000 --- a/framework/event_type.go +++ /dev/null @@ -1,32 +0,0 @@ -package framework - -type metaChange struct { - from uint64 - who taskRole - epoch uint64 - meta string -} - -type dataRequest struct { - taskID uint64 - epoch uint64 - req string - retry bool - dataChan chan []byte -} - -func (dr *dataRequest) notifyEpochMismatch() { - close(dr.dataChan) -} - -type dataResponse struct { - taskID uint64 - epoch uint64 - req string - data []byte - dataChan chan []byte -} - -func (dr *dataResponse) notifyEpochMismatch() { - close(dr.dataChan) -} diff --git a/framework/framework.go b/framework/framework.go deleted file mode 100644 index bc17166..0000000 --- a/framework/framework.go +++ /dev/null @@ -1,123 +0,0 @@ -package framework - -import ( - "fmt" - "log" - "math" - "net" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph/framework/frameworkhttp" - "github.com/taskgraph/taskgraph/pkg/etcdutil" -) - -const exitEpoch = math.MaxUint64 - -type framework struct { - // These should be passed by outside world - name string - etcdURLs []string - log *log.Logger - - // user defined interfaces - taskBuilder taskgraph.TaskBuilder - topology taskgraph.Topology - - task taskgraph.Task - taskID uint64 - epoch uint64 - etcdClient *etcd.Client - ln net.Listener - - // 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. - metaNotified map[string]bool - - // etcd stops - metaStops []chan bool - epochStop chan bool - - httpStop chan struct{} - heartbeatStop chan struct{} - - epochPassed chan struct{} - - // event loop - epochChan chan uint64 - metaChan chan *metaChange - dataReqtoSendChan chan *dataRequest - dataReqChan chan *dataRequest - dataRespToSendChan chan *dataResponse - dataRespChan chan *frameworkhttp.DataResponse -} - -func (f *framework) flagMetaToParent(meta string, epoch uint64) { - value := fmt.Sprintf("%d-%s", epoch, meta) - _, err := f.etcdClient.Set(etcdutil.ParentMetaPath(f.name, f.GetTaskID()), value, 0) - if err != nil { - f.log.Fatalf("etcdClient.Set failed; key: %s, value: %s, error: %v", - etcdutil.ParentMetaPath(f.name, f.GetTaskID()), value, err) - } -} - -func (f *framework) flagMetaToChild(meta string, epoch uint64) { - value := fmt.Sprintf("%d-%s", epoch, meta) - _, err := f.etcdClient.Set(etcdutil.ChildMetaPath(f.name, f.GetTaskID()), value, 0) - if err != nil { - f.log.Fatalf("etcdClient.Set failed; key: %s, value: %s, error: %v", - etcdutil.ChildMetaPath(f.name, f.GetTaskID()), value, err) - } -} - -// When app code invoke this method on framework, we simply -// update the etcd epoch to next uint64. All nodes should watch -// for epoch and update their local epoch correspondingly. -func (f *framework) incEpoch(epoch uint64) { - err := etcdutil.CASEpoch(f.etcdClient, f.name, epoch, epoch+1) - if err != nil { - f.log.Fatalf("task %d Epoch CompareAndSwap(%d, %d) failed: %v", - f.taskID, epoch+1, epoch, err) - } -} - -func (f *framework) dataRequest(toID uint64, req string, epoch uint64) { - // 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{ - taskID: toID, - epoch: epoch, - req: req, - } -} - -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) Kill() { - f.stop() -} - -// When node call this on framework, it simply set epoch to exitEpoch, -// All nodes will be notified of the epoch change and exit themselves. -func (f *framework) ShutdownJob() { - if err := etcdutil.CASEpoch(f.etcdClient, f.name, f.epoch, exitEpoch); err != nil { - panic("TODO: we should do a set instead of CAS here.") - } - if err := etcdutil.SetJobStatus(f.etcdClient, f.name, 0); err != nil { - panic("SetJobStatus") - } -} - -func (f *framework) GetLogger() *log.Logger { return f.log } - -func (f *framework) GetTaskID() uint64 { return f.taskID } - -func (f *framework) GetEpoch() uint64 { return f.epoch } diff --git a/framework/framework_test.go b/framework/framework_test.go deleted file mode 100644 index 999dc4b..0000000 --- a/framework/framework_test.go +++ /dev/null @@ -1,316 +0,0 @@ -package framework - -import ( - "net" - "reflect" - "sync" - "testing" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph" - "github.com/taskgraph/taskgraph/controller" - "github.com/taskgraph/taskgraph/example/topo" - "github.com/taskgraph/taskgraph/framework/frameworkhttp" - "github.com/taskgraph/taskgraph/pkg/etcdutil" -) - -// TestRequestDataEpochMismatch creates a scenario where data request happened -// with two different epochs. In this case, the server should back pressure and -// request client should get notified and return error. -func TestRequestDataEpochMismatch(t *testing.T) { - job := "TestRequestDataEpochMismatch" - etcdURLs := []string{"http://localhost:4001"} - ctl := controller.New(job, etcd.NewClient(etcdURLs), 1) - ctl.InitEtcdLayout() - defer ctl.DestroyEtcdLayout() - - fw := &framework{ - name: job, - etcdURLs: etcdURLs, - ln: createListener(t), - } - var wg sync.WaitGroup - fw.SetTaskBuilder(&testableTaskBuilder{ - setupLatch: &wg, - }) - fw.SetTopology(topo.NewTreeTopology(1, 1)) - wg.Add(1) - go fw.Start() - wg.Wait() - defer fw.ShutdownJob() - - addr, err := etcdutil.GetAddress(fw.etcdClient, job, fw.GetTaskID()) - if err != nil { - t.Fatalf("GetAddress failed: %v", err) - } - _, err = frameworkhttp.RequestData(addr, "req", 0, fw.GetTaskID(), 10, fw.GetLogger()) - if err != frameworkhttp.ErrReqEpochMismatch { - t.Fatalf("error want = %v, but get = (%)", frameworkhttp.ErrReqEpochMismatch, err.Error()) - } -} - -// TestFrameworkFlagMetaReady and TestFrameworkDataRequest test basic workflows of -// framework impl. It uses a scenario with two nodes: 0 as parent, 1 as child. -// The basic idea is that when parent tries to talk to child and vice versa, -// 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" - etcdURLs := []string{"http://localhost:4001"} - // launch controller to setup etcd layout - ctl := controller.New(appName, etcd.NewClient(etcdURLs), 2) - if err := ctl.InitEtcdLayout(); err != nil { - t.Fatalf("initEtcdLayout failed: %v", err) - } - defer ctl.DestroyEtcdLayout() - - pDataChan := make(chan *tDataBundle, 1) - cDataChan := make(chan *tDataBundle, 1) - - // simulate two tasks on two nodes -- 0 and 1 - // 0 is parent, 1 is child - f0 := &framework{ - name: appName, - etcdURLs: etcdURLs, - ln: createListener(t), - } - f1 := &framework{ - name: appName, - etcdURLs: etcdURLs, - ln: createListener(t), - } - - var wg sync.WaitGroup - taskBuilder := &testableTaskBuilder{ - dataMap: nil, - cDataChan: cDataChan, - pDataChan: pDataChan, - setupLatch: &wg, - } - f0.SetTaskBuilder(taskBuilder) - f0.SetTopology(topo.NewTreeTopology(2, 2)) - f1.SetTaskBuilder(taskBuilder) - f1.SetTopology(topo.NewTreeTopology(2, 2)) - - taskBuilder.setupLatch.Add(2) - go f0.Start() - go f1.Start() - taskBuilder.setupLatch.Wait() - if f0.GetTaskID() != 0 { - f0, f1 = f1, f0 - } - - defer f0.ShutdownJob() - - tests := []struct { - cMeta string - pMeta string - }{ - {"parent", "child"}, - {"ParamReady", "GradientReady"}, - } - - for i, tt := range tests { - // 0: F#FlagChildMetaReady -> 1: T#ParentMetaReady - f0.flagMetaToChild(tt.cMeta, 0) - // from child(1)'s view - data := <-pDataChan - expected := &tDataBundle{0, tt.cMeta, "", nil} - if !reflect.DeepEqual(data, expected) { - t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) - } - - // 1: F#FlagParentMetaReady -> 0: T#ChildMetaReady - f1.flagMetaToParent(tt.pMeta, 0) - // from parent(0)'s view - data = <-cDataChan - expected = &tDataBundle{1, tt.pMeta, "", nil} - if !reflect.DeepEqual(data, expected) { - t.Errorf("#%d: data bundle want = %v, get = %v", i, expected, data) - } - } -} - -func TestFrameworkDataRequest(t *testing.T) { - appName := "framework_test_flagmetaready" - etcdURLs := []string{"http://localhost:4001"} - // launch controller to setup etcd layout - ctl := controller.New(appName, etcd.NewClient(etcdURLs), 2) - if err := ctl.InitEtcdLayout(); err != nil { - t.Fatalf("initEtcdLayout failed: %v", err) - } - 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 - // 0 is parent, 1 is child - f0 := &framework{ - name: appName, - etcdURLs: etcdURLs, - ln: createListener(t), - } - f1 := &framework{ - name: appName, - etcdURLs: etcdURLs, - ln: createListener(t), - } - - var wg sync.WaitGroup - taskBuilder := &testableTaskBuilder{ - dataMap: dataMap, - cDataChan: cDataChan, - pDataChan: pDataChan, - setupLatch: &wg, - } - f0.SetTaskBuilder(taskBuilder) - f0.SetTopology(topo.NewTreeTopology(2, 2)) - f1.SetTaskBuilder(taskBuilder) - f1.SetTopology(topo.NewTreeTopology(2, 2)) - - taskBuilder.setupLatch.Add(2) - go f0.Start() - go f1.Start() - taskBuilder.setupLatch.Wait() - if f0.GetTaskID() != 0 { - f0, f1 = f1, f0 - } - - defer f0.ShutdownJob() - - for i, tt := range tests { - // 0: F#DataRequest -> 1: T#ServeAsChild -> 0: T#ChildDataReady - f0.dataRequest(1, tt.req, 0) - // 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(0, tt.req, 0) - // 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) - } - } -} - -type tDataBundle struct { - id uint64 - meta string - req string - resp []byte -} - -type testableTaskBuilder struct { - dataMap map[string][]byte - cDataChan chan *tDataBundle - pDataChan chan *tDataBundle - setupLatch *sync.WaitGroup -} - -func (b *testableTaskBuilder) GetTask(taskID uint64) taskgraph.Task { - switch taskID { - case 0: - return &testableTask{dataMap: b.dataMap, dataChan: b.cDataChan, - setupLatch: b.setupLatch} - case 1: - return &testableTask{dataMap: b.dataMap, dataChan: b.pDataChan, - setupLatch: b.setupLatch} - default: - panic("unimplemented") - } -} - -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. - // - // The basic idea is that there are only two nodes -- one parent and one child. - // When this channel is for parent, it passes information from child. - dataChan chan *tDataBundle -} - -func (t *testableTask) Init(taskID uint64, framework taskgraph.Framework) { - t.id = taskID - t.framework = framework - if t.setupLatch != nil { - t.setupLatch.Done() - } -} -func (t *testableTask) Exit() {} -func (t *testableTask) SetEpoch(ctx taskgraph.Context, epoch uint64) {} - -func (t *testableTask) ParentMetaReady(ctx taskgraph.Context, fromID uint64, meta string) { - if t.dataChan != nil { - t.dataChan <- &tDataBundle{fromID, meta, "", nil} - } -} - -func (t *testableTask) ChildMetaReady(ctx taskgraph.Context, fromID uint64, meta string) { - t.ParentMetaReady(ctx, fromID, meta) -} - -func (t *testableTask) ServeAsParent(fromID uint64, req string, dataReceiver chan<- []byte) { - if t.dataChan != nil { - t.dataChan <- &tDataBundle{fromID, "", req, nil} - } - dataReceiver <- t.dataMap[req] -} -func (t *testableTask) ServeAsChild(fromID uint64, req string, dataReceiver chan<- []byte) { - t.ServeAsParent(fromID, req, dataReceiver) -} -func (t *testableTask) ParentDataReady(ctx taskgraph.Context, fromID uint64, req string, resp []byte) { - if t.dataChan != nil { - t.dataChan <- &tDataBundle{fromID, "", req, resp} - } -} - -func (t *testableTask) ChildDataReady(ctx taskgraph.Context, fromID uint64, req string, resp []byte) { - t.ParentDataReady(ctx, fromID, req, resp) -} - -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 -} diff --git a/framework/frameworkhttp/data_request_handler.go b/framework/frameworkhttp/data_request_handler.go deleted file mode 100644 index 60236f5..0000000 --- a/framework/frameworkhttp/data_request_handler.go +++ /dev/null @@ -1,124 +0,0 @@ -package frameworkhttp - -import ( - "errors" - "io/ioutil" - "log" - "net/http" - "net/url" - "strconv" -) - -var ( - ErrReqEpochMismatch error = errors.New("data request error: epoch mismatch") - ErrServerClosed error = errors.New("server has been closed") -) - -const ( - DataRequestPrefix string = "/datareq" - DataRequestTaskID string = "taskID" - DataRequestReq string = "req" - DataRequestEpoch string = "epoch" -) - -type DataGetter interface { - GetTaskData(uint64, uint64, string) ([]byte, error) -} - -type dataReqHandler struct { - logger *log.Logger - DataGetter -} - -type DataResponse struct { - TaskID uint64 - Epoch uint64 - Req string - Data []byte -} - -func NewDataRequestHandler(logger *log.Logger, dg DataGetter) http.Handler { - return &dataReqHandler{ - logger: logger, - DataGetter: dg, - } -} - -func (h *dataReqHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != DataRequestPrefix { - http.Error(w, "bad path", http.StatusBadRequest) - return - } - // parse url query - q := r.URL.Query() - fromIDStr := q.Get(DataRequestTaskID) - fromID, err := strconv.ParseUint(fromIDStr, 0, 64) - if err != nil { - h.logger.Panic("Internal error: fromID couldn't be parsed") - } - epochStr := q.Get(DataRequestEpoch) - epoch, err := strconv.ParseUint(epochStr, 0, 64) - if err != nil { - h.logger.Panic("Internal error: epoch couldn't be parsed") - } - req := q.Get(DataRequestReq) - - b, err := h.GetTaskData(fromID, epoch, req) - if err != nil { - if err == ErrReqEpochMismatch || err == ErrServerClosed { - w.WriteHeader(http.StatusInternalServerError) - w.Write([]byte(err.Error())) - return - } - h.logger.Panic("unimplemented") - } - if _, err := w.Write(b); err != nil { - log.Printf("http: response write failed: %v", err) - } -} - -func RequestData(addr string, req string, from, to, epoch uint64, logger *log.Logger) (*DataResponse, error) { - u := url.URL{ - Scheme: "http", - Host: addr, - Path: DataRequestPrefix, - } - q := u.Query() - q.Add(DataRequestTaskID, strconv.FormatUint(from, 10)) - q.Add(DataRequestReq, req) - q.Add(DataRequestEpoch, strconv.FormatUint(epoch, 10)) - u.RawQuery = q.Encode() - urlStr := u.String() - // send request - // pass the response to the awaiting event loop for data response - resp, err := http.Get(urlStr) - if err != nil { - // The error could be caused because: 1. network failure; 2. We might have - // sent request to failed server. - return nil, err - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusInternalServerError { - logger.Fatalf("http: response code = %d, expect = %d", resp.StatusCode, 200) - } - data, err := ioutil.ReadAll(resp.Body) - if err != nil { - logger.Fatalf("http: ioutil.ReadAll(%v) returns error: %v", resp.Body, err) - } - if resp.StatusCode == http.StatusInternalServerError { - logger.Printf("http error bytes: %s", string(data)) - if string(data) == ErrReqEpochMismatch.Error() { - return nil, ErrReqEpochMismatch - } - if string(data) == ErrServerClosed.Error() { - return nil, ErrServerClosed - } - logger.Fatalf("Unknown error: %v", data) - } - return &DataResponse{ - TaskID: to, - Epoch: epoch, - Req: req, - Data: data, - }, nil -} diff --git a/framework/healthy.go b/framework/healthy.go deleted file mode 100644 index b269496..0000000 --- a/framework/healthy.go +++ /dev/null @@ -1,21 +0,0 @@ -package framework - -import ( - "time" - - "github.com/taskgraph/taskgraph/pkg/etcdutil" -) - -var ( - heartbeatInterval = 1 * time.Second -) - -func (f *framework) heartbeat() { - f.heartbeatStop = make(chan struct{}) - go func() { - err := etcdutil.Heartbeat(f.etcdClient, f.name, f.taskID, heartbeatInterval, f.heartbeatStop) - if err != nil { - f.log.Printf("Heartbeat stops with error: %v\n", err) - } - }() -} diff --git a/framework_interface.go b/framework_interface.go index f65b347..fc30768 100644 --- a/framework_interface.go +++ b/framework_interface.go @@ -1,61 +1,56 @@ package taskgraph -import "log" +import "github.com/taskgraph/taskgraph/job" -// This interface is used by application during taskgraph configuration phase. +// This interface is used by application to configure task builder and +// do bootstrap start. type Bootstrap interface { - // These allow application developer to set the task configuration so framework - // implementation knows which task to invoke at each node. - SetTaskBuilder(taskBuilder TaskBuilder) - - // This allow the application to specify how tasks are connection at each epoch - SetTopology(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. + SetTaskBuilder(TaskBuilder) Start() } -// Framework hides distributed system complexity and provides users convenience of -// high level features. type Framework interface { - // This allow the task implementation query its neighbors. - GetTopology() Topology - // Kill the framework itself. - // As epoch changes, some nodes isn't needed anymore + // As epoch changes, some tasks isn't needed anymore. Kill() - // Some task can inform all participating tasks to shutdown. - // If successful, all tasks will be gracefully shutdown. - // TODO: @param status - ShutdownJob() + // Inform all participating tasks to shutdown. + // All tasks will be gracefully shutdown. + ShutdownJob(es job.ExitStatus) - GetLogger() *log.Logger - - // This is used to figure out taskid for current node - GetTaskID() uint64 + // TaskID for current node + TaskID() uint64 } -// Context is used in task callbacks. It provides APIs for tasks to ask framework -// to do work in certain context. type Context interface { - // These two are useful for task to inform the framework their status change. - // metaData has to be really small, since it might be stored in etcd. - // Set meta flag to notify parent/child of the change. - FlagMetaToParent(meta string) - FlagMetaToChild(meta string) + CreateComposer() Composer +} - // Some task can inform all participating tasks to new epoch - IncEpoch() +type Composer interface { + SetJoint(Joint) + CreateInboundChannel(taskID uint64, tag string) + CreateOutboundChannel(taskID uint64, tag string) +} + +type channelBasics interface { + TaskID() uint64 + Tag() string +} - // Request data from parent or children. - DataRequest(toID uint64, meta string) +type InboundChannel interface { + channelBasics + Get() []byte } -// Note that framework can decide how update can be done, and how to serve the updatelog. -type BackedUpFramework interface { - // Ask framework to do update on this update on this task, which consists - // of one primary and some backup copies. - Update(taskID uint64, log UpdateLog) +type OutboundChannel interface { + channelBasics + Put(Marshaler) } + +// TODO: +// 1. User proto message (deferred) +// 2. Channel Implemented in gRPC +// - Inbound has a client. Client gets data and then dispatches to all inbound channels +// that requested. +// - Outbound has a server. +// 3. Run regression diff --git a/integration/node_failure_test.go b/integration/node_failure_test.go deleted file mode 100644 index 1004b8e..0000000 --- a/integration/node_failure_test.go +++ /dev/null @@ -1,121 +0,0 @@ -package integration - -import ( - "log" - "testing" - - "github.com/coreos/go-etcd/etcd" - "github.com/taskgraph/taskgraph/controller" - "github.com/taskgraph/taskgraph/example/regression" -) - -// TestMasterSetEpochFailure checks if a master task failed at SetEpoch, -// 1. a new boostrap will be created to take over -// 2. continue what's left; -// 3. finish the job with the same result. -func TestMasterSetEpochFailure(t *testing.T) { - job := "TestMasterSetEpochFailure" - etcdURLs := []string{"http://localhost:4001"} - numOfTasks := uint64(15) - numOfIterations := uint64(10) - - // controller start first to setup task directories in etcd - controller := controller.New(job, etcd.NewClient(etcdURLs), numOfTasks) - controller.Start() - - taskBuilder := ®ression.SimpleTaskBuilder{ - GDataChan: make(chan int32, 11), - NodeProducer: make(chan bool, 1), - MasterConfig: map[string]string{ - "SetEpoch": "fail", - "failepoch": "1", - "faillevel": "100", - }, - NumberOfIterations: numOfIterations, - } - for i := uint64(0); i < numOfTasks; i++ { - go drive(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) - } - - wantData := []int32{0, 105, 210, 315, 420, 525, 630, 735, 840, 945, 1050} - getData := make([]int32, numOfIterations+1) - for i := uint64(0); i <= numOfIterations; i++ { - getData[i] = <-taskBuilder.GDataChan - } - - for i := range wantData { - if wantData[i] != getData[i] { - t.Errorf("#%d: data want = %d, get = %d", i, wantData[i], getData[i]) - } - } - controller.WaitForJobDone() - controller.Stop() -} - -func TestSlaveParentDataReadyFailure(t *testing.T) { - job := "TestSlavePDataReadyFailure" - slaveConfig := map[string]string{ - "ParentDataReady": "fail", - "faillevel": "3", - } - testSlaveFailure(t, job, slaveConfig) -} - -// This test tests fault tolerance in slave ChildDataReady() if node fails before/after -// sending data to parent node -func TestSlaveChildDataReadyFailure(t *testing.T) { - job := "TestSlaveChildDataReadyFailure" - slaveConfig := map[string]string{ - "ChildDataReady": "fail", - "faillevel": "3", - } - testSlaveFailure(t, job, slaveConfig) -} - -func testSlaveFailure(t *testing.T, job string, slaveConfig map[string]string) { - etcdURLs := []string{"http://localhost:4001"} - numOfTasks := uint64(15) - numOfIterations := uint64(10) - - // controller start first to setup task directories in etcd - controller := controller.New(job, etcd.NewClient(etcdURLs), numOfTasks) - controller.Start() - defer controller.Stop() - - taskBuilder := ®ression.SimpleTaskBuilder{ - GDataChan: make(chan int32, 11), - NodeProducer: make(chan bool, 1), - SlaveConfig: slaveConfig, - NumberOfIterations: numOfIterations, - } - go func() { - for _ = range taskBuilder.NodeProducer { - log.Println("Starting a new node") - go drive(t, job, etcdURLs, numOfTasks, taskBuilder) - } - }() - for i := uint64(0); i < numOfTasks; i++ { - taskBuilder.NodeProducer <- true - } - - wantData := []int32{0, 105, 210, 315, 420, 525, 630, 735, 840, 945, 1050} - getData := make([]int32, numOfIterations+1) - for i := uint64(0); i <= numOfIterations; i++ { - getData[i] = <-taskBuilder.GDataChan - } - - for i := range wantData { - if wantData[i] != getData[i] { - t.Errorf("#%d: data want = %d, get = %d", i, wantData[i], getData[i]) - } - } - controller.WaitForJobDone() - controller.Stop() - close(taskBuilder.NodeProducer) -} diff --git a/integration/regression_framework_test.go b/integration/regression_framework_test.go deleted file mode 100644 index ab4e917..0000000 --- a/integration/regression_framework_test.go +++ /dev/null @@ -1,64 +0,0 @@ -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) { - etcdURLs := []string{"http://localhost:4001"} - - job := "framework_regression_test" - numOfTasks := uint64(15) - numOfIterations := uint64(10) - - // controller start first to setup task directories in etcd - controller := controller.New(job, etcd.NewClient(etcdURLs), numOfTasks) - controller.Start() - - // We need to set etcd so that nodes know what to do. - taskBuilder := ®ression.SimpleTaskBuilder{ - GDataChan: make(chan int32, 11), - NumberOfIterations: numOfIterations, - } - for i := uint64(0); i < numOfTasks; i++ { - go drive(t, job, etcdURLs, numOfTasks, taskBuilder) - } - - wantData := []int32{0, 105, 210, 315, 420, 525, 630, 735, 840, 945, 1050} - getData := make([]int32, numOfIterations+1) - for i := uint64(0); i <= numOfIterations; i++ { - getData[i] = <-taskBuilder.GDataChan - } - for i := range wantData { - if wantData[i] != getData[i] { - t.Errorf("#%d: data want = %d, get = %d\n", i, wantData[i], getData[i]) - } - } - - 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/job/exit_status.go b/job/exit_status.go new file mode 100644 index 0000000..be078a9 --- /dev/null +++ b/job/exit_status.go @@ -0,0 +1,8 @@ +package job + +const ( + SuccessStatus ExitStatus = iota + 1 + FailureStatus +) + +type ExitStatus int diff --git a/node_interface.go b/node_interface.go deleted file mode 100644 index 6a5cac7..0000000 --- a/node_interface.go +++ /dev/null @@ -1,16 +0,0 @@ -package taskgraph - -type Node interface { - // return the ID of this node - ID() uint64 - // return the task this node associated to - TaskID() uint64 - // return the status of this node - // possible status: no associated to any task - // master of a task - // slave of a task - Status() uint64 - // return a connection string of this node - // scheme://host:port - Connection() string -} diff --git a/pkg/common/countdown_latch.go b/pkg/common/countdown_latch.go deleted file mode 100644 index b01e1d2..0000000 --- a/pkg/common/countdown_latch.go +++ /dev/null @@ -1,45 +0,0 @@ -package common - -import "sync" - -// I am writing this count down latch because sync.WaitGroup doesn't support -// decrementing counter when it's 0. -type CountdownLatch struct { - sync.Mutex - cond *sync.Cond - counter int -} - -func NewCountdownLatch(count int) *CountdownLatch { - c := new(CountdownLatch) - c.cond = sync.NewCond(c) - c.counter = count - return c -} - -func (c *CountdownLatch) Count() int { - c.Lock() - defer c.Unlock() - return c.counter -} - -func (c *CountdownLatch) CountDown() { - c.Lock() - defer c.Unlock() - if c.counter == 0 { - return - } - c.counter-- - if c.counter == 0 { - c.cond.Broadcast() - } -} - -func (c *CountdownLatch) Await() { - c.Lock() - defer c.Unlock() - if c.counter == 0 { - return - } - c.cond.Wait() -} diff --git a/pkg/etcdutil/epoch.go b/pkg/etcdutil/epoch.go deleted file mode 100644 index b4ea006..0000000 --- a/pkg/etcdutil/epoch.go +++ /dev/null @@ -1,42 +0,0 @@ -package etcdutil - -import ( - "log" - "strconv" - - "github.com/coreos/go-etcd/etcd" -) - -func GetAndWatchEpoch(client *etcd.Client, appname string, epochC chan uint64, stop chan bool) (uint64, error) { - resp, err := client.Get(EpochPath(appname), false, false) - if err != nil { - log.Fatal("etcdutil: can not get epoch from etcd") - } - ep, err := strconv.ParseUint(resp.Node.Value, 10, 64) - if err != nil { - return 0, err - } - receiver := make(chan *etcd.Response, 1) - go client.Watch(EpochPath(appname), resp.EtcdIndex+1, false, receiver, stop) - go func() { - for resp := range receiver { - if resp.Action != "compareAndSwap" && resp.Action != "set" { - continue - } - epoch, err := strconv.ParseUint(resp.Node.Value, 10, 64) - if err != nil { - log.Fatal("etcdutil: can't parse epoch from etcd") - } - epochC <- epoch - } - }() - - return ep, nil -} - -func CASEpoch(client *etcd.Client, appname string, prevEpoch, epoch uint64) error { - prevEpochStr := strconv.FormatUint(prevEpoch, 10) - epochStr := strconv.FormatUint(epoch, 10) - _, err := client.CompareAndSwap(EpochPath(appname), epochStr, 0, prevEpochStr, 0) - return err -} diff --git a/pkg/etcdutil/healthy.go b/pkg/etcdutil/healthy.go deleted file mode 100644 index 1b8ff69..0000000 --- a/pkg/etcdutil/healthy.go +++ /dev/null @@ -1,105 +0,0 @@ -package etcdutil - -import ( - "fmt" - "log" - "math/rand" - "path" - "strconv" - "time" - - "github.com/coreos/go-etcd/etcd" -) - -// heartbeat to etcd cluster until stop -func Heartbeat(client *etcd.Client, name string, taskID uint64, interval time.Duration, stop chan struct{}) error { - for { - _, err := client.Set(TaskHealthyPath(name, taskID), "health", computeTTL(interval)) - if err != nil { - return err - } - select { - case <-time.After(interval): - case <-stop: - return nil - } - } -} - -// detect failure of the given taskID -func DetectFailure(client *etcd.Client, name string, stop chan bool) error { - receiver := make(chan *etcd.Response, 1) - go client.Watch(HealthyPath(name), 0, true, receiver, stop) - for resp := range receiver { - if resp.Action != "expire" && resp.Action != "delete" { - continue - } - if err := ReportFailure(client, name, path.Base(resp.Node.Key)); err != nil { - return err - } - } - return nil -} - -// report failure to etcd cluster -// If a framework detects a failure, it tries to report failure to /FreeTasks/{taskID} -func ReportFailure(client *etcd.Client, name, failedTask string) error { - _, err := client.Set(FreeTaskPath(name, failedTask), "failed", 0) - return err -} - -// WaitFreeTask blocks until it gets a hint of free task -func WaitFreeTask(client *etcd.Client, name string, logger *log.Logger) (uint64, error) { - slots, err := client.Get(FreeTaskDir(name), false, true) - if err != nil { - return 0, err - } - if total := len(slots.Node.Nodes); total > 0 { - ri := rand.Intn(total) - s := slots.Node.Nodes[ri] - idStr := path.Base(s.Key) - id, err := strconv.ParseUint(idStr, 0, 64) - if err != nil { - return 0, err - } - logger.Printf("got free task %v, randomly choose %d to try...", ListKeys(slots.Node.Nodes), ri) - return id, nil - } - - watchIndex := slots.EtcdIndex + 1 - respChan := make(chan *etcd.Response, 1) - go func() { - for { - logger.Printf("start to wait failure at index %d", watchIndex) - resp, err := client.Watch(FreeTaskDir(name), watchIndex, true, nil, nil) - if err != nil { - logger.Printf("WARN: WaitFailure watch failed: %v", err) - return - } - if resp.Action == "set" { - respChan <- resp - return - } - watchIndex = resp.EtcdIndex + 1 - } - }() - 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 - } - return id, nil -} - -func computeTTL(interval time.Duration) uint64 { - if interval/time.Second < 1 { - return 3 - } - return 3 * uint64(interval/time.Second) -} diff --git a/pkg/etcdutil/layout.go b/pkg/etcdutil/layout.go deleted file mode 100644 index bad1f6a..0000000 --- a/pkg/etcdutil/layout.go +++ /dev/null @@ -1,80 +0,0 @@ -package etcdutil - -import ( - "path" - "strconv" -) - -// The directory layout we going to define in etcd: -// /{app}/config -> application configuration -// /{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}/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} - -const ( - TasksDir = "tasks" - NodesDir = "nodes" - ConfigDir = "config" - FreeDir = "freeTasks" - Epoch = "epoch" - Status = "status" - TaskMaster = "0" - TaskParentMeta = "parentMeta" - TaskChildMeta = "childMeta" - NodeAddr = "address" - NodeTTL = "ttl" - Healthy = "healthy" -) - -func EpochPath(appName string) string { - return path.Join("/", appName, Epoch) -} - -func JobStatusPath(appName string) string { - return path.Join("/", appName, Status) -} - -func HealthyPath(appName string) string { - return path.Join("/", appName, Healthy) -} - -func TaskHealthyPath(appName string, taskID uint64) string { - return path.Join(HealthyPath(appName), strconv.FormatUint(taskID, 10)) -} -func FreeTaskDir(appName string) string { - return path.Join("/", appName, FreeDir) -} -func FreeTaskPath(appName, idStr string) string { - return path.Join(FreeTaskDir(appName), idStr) -} - -func TaskDirPath(appName string) string { - return path.Join("/", appName, TasksDir) -} - -func TaskMasterPath(appName string, taskID uint64) string { - return path.Join("/", appName, TasksDir, strconv.FormatUint(taskID, 10), TaskMaster) -} - -func ParentMetaPath(appName string, taskID uint64) string { - return path.Join("/", - appName, - TasksDir, - strconv.FormatUint(taskID, 10), - TaskParentMeta) -} - -func ChildMetaPath(appName string, taskID uint64) string { - return path.Join("/", - appName, - TasksDir, - strconv.FormatUint(taskID, 10), - TaskChildMeta) -} diff --git a/pkg/etcdutil/meta.go b/pkg/etcdutil/meta.go deleted file mode 100644 index 71d208b..0000000 --- a/pkg/etcdutil/meta.go +++ /dev/null @@ -1,22 +0,0 @@ -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 { - 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) - } - 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) - } - }(receiver) - return nil -} diff --git a/pkg/etcdutil/task.go b/pkg/etcdutil/task.go deleted file mode 100644 index a327f2d..0000000 --- a/pkg/etcdutil/task.go +++ /dev/null @@ -1,39 +0,0 @@ -package etcdutil - -import ( - "log" - "strconv" - - "github.com/coreos/go-etcd/etcd" -) - -func TryOccupyTask(client *etcd.Client, name string, taskID uint64, connection string) bool { - _, err := client.Create(TaskHealthyPath(name, taskID), "health", 3) - if err != nil { - return false - } - 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 true -} - -// getAddress will return the host:port address of the service taking care of -// the task that we want to talk to. -// Currently we grab the information from etcd every time. Local cache could be used. -// If it failed, e.g. network failure, it should return error. -func GetAddress(client *etcd.Client, name string, id uint64) (string, error) { - resp, err := client.Get(TaskMasterPath(name, id), false, false) - if err != nil { - return "", err - } - return resp.Node.Value, nil -} - -func SetJobStatus(client *etcd.Client, name string, status int) error { - _, err := client.Set(JobStatusPath(name), "done", 0) - return err -} diff --git a/pkg/etcdutil/util.go b/pkg/etcdutil/util.go deleted file mode 100644 index e15e5dd..0000000 --- a/pkg/etcdutil/util.go +++ /dev/null @@ -1,23 +0,0 @@ -package etcdutil - -import ( - "log" - - "github.com/coreos/go-etcd/etcd" -) - -func ListKeys(nodes []*etcd.Node) []string { - res := make([]string, len(nodes)) - for i, n := range nodes { - res[i] = n.Key - } - return res -} - -func MustCreate(c *etcd.Client, logger *log.Logger, key, value string, ttl uint64) *etcd.Response { - resp, err := c.Create(key, value, ttl) - if err != nil { - logger.Panicf("Create failed. Key: %s, err: %v", key, err) - } - return resp -} diff --git a/pkg/topoutil/task_role.go b/pkg/topoutil/task_role.go deleted file mode 100644 index 9945346..0000000 --- a/pkg/topoutil/task_role.go +++ /dev/null @@ -1,21 +0,0 @@ -package topoutil - -import "github.com/taskgraph/taskgraph" - -func IsParent(t taskgraph.Topology, epoch, taskID uint64) bool { - for _, id := range t.GetParents(epoch) { - if taskID == id { - return true - } - } - return false -} - -func IsChild(t taskgraph.Topology, epoch, taskID uint64) bool { - for _, id := range t.GetChildren(epoch) { - if taskID == id { - return true - } - } - return false -} diff --git a/task_builder_interface.go b/task_builder_interface.go deleted file mode 100644 index 473a840..0000000 --- a/task_builder_interface.go +++ /dev/null @@ -1,12 +0,0 @@ -/* -TaskBuilder is also implemented by application developer and used by -framework implementation so decide which task implementation one should use -at any give node. It should be called only once at node initialization. -*/ -package taskgraph - -type TaskBuilder interface { - // This method is called once by framework implementation to get the - // right task implementation for given node/task. - GetTask(taskID uint64) Task -} diff --git a/task_interface.go b/task_interface.go index 25904d7..d69a126 100644 --- a/task_interface.go +++ b/task_interface.go @@ -1,45 +1,30 @@ package taskgraph -// Task is a logic repersentation of a computing unit. -// Each task contain at least one Node. -// Each task has exact one master Node and might have multiple salve Nodes. +// TaskBuilder should be implemented by application developer and used by +// framework implementation to decide which task implementation to be used +// at given node. +type TaskBuilder interface { + // This method is called once by framework implementation to get the + // right task implementation for given node. + Build(taskID uint64) Task +} -// All event handler functions and should be non-blocking. type Task interface { - // This is useful to bring the task up to speed from scratch or if it recovers. - Init(taskID uint64, framework Framework) - - // Task is finished up for exit. Last chance to save some task specific work. + // numberOfTasks: how many tasks are created for this job. + // User can use this number to make decision on topology. + Init(framework Framework, numberOfTasks uint64) Exit() // Framework tells user task what current epoch is. - // This give the task an opportunity to cleanup and regroup. + // User can compose a graph using channels and joints here. SetEpoch(ctx 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. - ParentMetaReady(ctx Context, parentID uint64, meta string) - ChildMetaReady(ctx Context, childID uint64, meta string) - ParentDataReady(ctx Context, parentID uint64, req string, resp []byte) - ChildDataReady(ctx Context, childID uint64, req string, resp []byte) - - // These are payload for application purpose. - ServeAsParent(fromID uint64, req string, dataReceiver chan<- []byte) - ServeAsChild(fromID uint64, req string, dataReceiver chan<- []byte) } -type UpdateLog interface { - UpdateID() +// user-implemented data processing/computing unit. +type Joint interface { + Compute(ins []InboundChannel, outs []OutboundChannel) } -// Backupable is an interface that task need to implement if they want to have -// hot standby copy. This is another can of beans. -type Backupable interface { - // Some hooks that need for master slave etc. - BecamePrimary() - BecameBackup() - - // Framework notify this copy to update. This should be the only way that - // one update the state of copy. - Update(log UpdateLog) +type Marshaler interface { + Marshal() ([]byte, error) } diff --git a/topology_interface.go b/topology_interface.go deleted file mode 100644 index 4793dd7..0000000 --- a/topology_interface.go +++ /dev/null @@ -1,29 +0,0 @@ -/* -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. 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 GetParents and GetChildren with given epoch, so that it knows - how to setup watcher for node failures. -*/ -package taskgraph - -// The Topology will be implemented by the application. -// Each Topology might have many epochs. The topology of each epoch -// might be different. -type Topology interface { - // This method is called once by framework implementation. So that - // we can get the local topology for each epoch later. - SetTaskID(taskID uint64) - - // GetParents returns the parents' IDs of this task at the - // given epoch. - GetParents(epoch uint64) []uint64 - - // GetChlidren returns the children's IDs of this task at the - // given epoch. - GetChildren(epoch uint64) []uint64 -}