scheduler

Documentation

Overview

Schedule tasks to run on available resources assigned by master.

Index

• func SendCleanupRequest(server string, request *pb.CleanupRequest) error
• type DatasetShardLocator
  • func NewDatasetShardLocator() *DatasetShardLocator
  • func (l *DatasetShardLocator) GetShardLocation(shardName string) (pb.DataLocation, bool)
  • func (l *DatasetShardLocator) SetShardLocation(name string, location pb.DataLocation)
• type Option
• type RemoteExecutorStatus
• type Scheduler
  • func New(leader string, option *Option) *Scheduler
  • func (s *Scheduler) DeleteOutput(taskGroup *plan.TaskGroup)
  • func (s *Scheduler) ExecuteTaskGroup(ctx context.Context, fc *flow.Flow, ...)
  • func (s *Scheduler) Fetch(demands []market.Demand)
  • func (s *Scheduler) GetShardLocation(shard *flow.DatasetShard) (pb.DataLocation, bool)
  • func (s *Scheduler) Score(r market.Requirement, bid float64, obj market.Object) float64

Functions

func SendCleanupRequest

func SendCleanupRequest(server string, request *pb.CleanupRequest) error

Types

type DatasetShardLocator

type DatasetShardLocator struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewDatasetShardLocator

func NewDatasetShardLocator() *DatasetShardLocator

func (*DatasetShardLocator) GetShardLocation

func (l *DatasetShardLocator) GetShardLocation(shardName string) (pb.DataLocation, bool)

func (*DatasetShardLocator) SetShardLocation

func (l *DatasetShardLocator) SetShardLocation(name string, location pb.DataLocation)

type Option

type Option struct {
	Username     string
	Hostname     string
	FlowHashcode uint32
	DataCenter   string
	Rack         string
	TaskMemoryMB int
	Module       string
	IsProfiling  bool
}

type RemoteExecutorStatus

type RemoteExecutorStatus struct {
	Request      *pb.ExecutionRequest
	Allocation   *pb.Allocation
	RequestTime  time.Time
	InputLength  int
	OutputLength int
	ReadyTime    time.Time
	RunTime      time.Time
	StopTime     time.Time
}

type Scheduler

type Scheduler struct {
	sync.Mutex

	Master    string
	EventChan chan interface{}
	Market    *market.Market
	Option    *Option
	// contains filtered or unexported fields
}

func New

func New(leader string, option *Option) *Scheduler

func (*Scheduler) DeleteOutput

func (s *Scheduler) DeleteOutput(taskGroup *plan.TaskGroup)

func (*Scheduler) ExecuteTaskGroup

func (s *Scheduler) ExecuteTaskGroup(ctx context.Context,
	fc *flow.Flow,
	taskGroupStatus *pb.FlowExecutionStatus_TaskGroup,
	wg *sync.WaitGroup,
	taskGroup *plan.TaskGroup,
	bid float64, relatedFiles []resource.FileResource, binaryPath string)

ExecuteTaskGroup wait for inputs and execute the task group remotely. If cancelled, the output will be cleaned up.

func (*Scheduler) Fetch

func (s *Scheduler) Fetch(demands []market.Demand)

Requirement is TaskGroup Object is Agent's Location

func (*Scheduler) GetShardLocation

func (s *Scheduler) GetShardLocation(shard *flow.DatasetShard) (pb.DataLocation, bool)

func (*Scheduler) Score

func (s *Scheduler) Score(r market.Requirement, bid float64, obj market.Object) float64