taskgraph

README

TaskGraph

TaskGraph is a framework for writing fault tolerent distributed machine learning 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 (typed neighbors fail/restart, primary/backup switch) to task implementation so that it can do application dependent recovery.

TaskGrpah supports an event driven pull model for communication. When one task want to talk to some other task, it set a flag via framework, and framework will notify recipient, which can decide whether or when to fetch data via Framework. Framework will handle communication failures via automatic retry, reconnect to new task node, etc. In another word, it provides exactly-once semantic on data request.

An TaskGraph application usually has three layers. And application implementation need to configure TaskGraph in driver layer and also implement Task/TaskBuilder/Topology based on application logic.

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 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 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 example/regression.

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.

Documentation

Overview

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.

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 GetLinkTypes and GetNeighbors with given epoch, so that it knows
how to setup watcher for node failures.

Index

• type BackedUpFramework
• type Backupable
• type Bootstrap
• type Datum
• type DatumIterator
• type DatumIteratorBuilder
• type DatumStore
• type DatumTransformer
• type Framework
• type Node
• type Task
• type TaskBuilder
• type Topology
• type UpdateLog

Types

type BackedUpFramework

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

Note that framework can decide how update can be done, and how to serve the updatelog.

type Backupable

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

Backupable is an interface that task need to implement if they want to have hot standby copy. This is another can of beans.

type Bootstrap

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.
	Start()
}

This interface is used by application during taskgraph configuration phase.

type Datum

type Datum interface{}

Datum is the interface for basic loading and transformation.

type DatumIterator

type DatumIterator interface {
	HasNext() bool
	Next() Datum
}

DatumIerator allow one to iterate through all the datum in the set.

type DatumIteratorBuilder

type DatumIteratorBuilder interface {
	Build(path string) DatumIterator
}

This can be used to build a sequence of Datum from source.

type DatumStore

type DatumStore struct {
	Cache []Datum
}

DatumStore host a set of Datum in the memory.

type DatumTransformer

type DatumTransformer interface {
	Transform(old Datum) Datum
}

Transform Datum from one format to another.

type Framework

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
	Kill()

	// Some task can inform all participating tasks to shutdown.
	// If successful, all tasks will be gracefully shutdown.
	// TODO: @param status
	ShutdownJob()

	GetLogger() *log.Logger

	// This is used to figure out taskid for current node
	GetTaskID() uint64

	// This is 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 meta to all nodes of linkType to this node.
	FlagMeta(ctx context.Context, linkType, meta string)

	// Some task can inform all participating tasks to new epoch
	IncEpoch(ctx context.Context)

	// Request data from task toID with specified linkType and meta.
	DataRequest(ctx context.Context, toID uint64, method string, input proto.Message)
	CheckGRPCContext(ctx context.Context) error
}

Framework hides distributed system complexity and provides users convenience of high level features.

type Node

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
}

type Task

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.
	Exit()

	// Framework tells user task what current epoch is.
	// This give the task an opportunity to cleanup and regroup.
	EnterEpoch(ctx context.Context, epoch uint64)

	// The meta/data notifications obey exactly-once semantics. Note that the same
	// meta string will be notified only once even if you flag the meta more than once.
	// TODO: one can also get this from channel.
	MetaReady(ctx context.Context, fromID uint64, linkType, meta string)

	// This is the callback when data from server is ready.
	DataReady(ctx context.Context, fromID uint64, method string, output proto.Message)

	CreateOutputMessage(methodName string) proto.Message
	CreateServer() *grpc.Server
}

All event handler functions and should be non-blocking.

type TaskBuilder

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
}

type Topology

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)

	// This returns the type of links this topology supports
	GetLinkTypes() []string

	// This returns the neighbors of given link for this node at this epoch.
	GetNeighbors(linkType string, epoch uint64) []uint64
}

The Topology will be implemented by the application. Each Topology might have many epochs. The topology of each epoch might be different.

type UpdateLog

type UpdateLog interface {
	UpdateID()
}