task

package
v2.7.0-nightly.20230519 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 18, 2023 License: Apache-2.0 Imports: 32 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidLengthTask        = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowTask          = fmt.Errorf("proto: integer overflow")
	ErrUnexpectedEndOfGroupTask = fmt.Errorf("proto: unexpected end of group")
)
View Source 
var State_name = map[int32]string{
	0: "STATE_UNKNOWN",
	1: "RUNNING",
	2: "SUCCESS",
	3: "FAILURE",
}
View Source 
var State_value = map[string]int32{
	"STATE_UNKNOWN": 0,
	"RUNNING":       1,
	"SUCCESS":       2,
	"FAILURE":       3,
}

Functions

func Count 

func Count(ctx context.Context, service Service, namespace, group string) (tasks int64, claims int64, retErr error)

Count returns the number of tasks and claims in the given namespace and group (if nonempty)

func DoBatch 

func DoBatch(ctx context.Context, doer Doer, inputs []*types.Any, cb CollectFunc) error

DoBatch executes a batch of tasks.

func DoOne 

func DoOne(ctx context.Context, doer Doer, input *types.Any) (*types.Any, error)

DoOne executes one task. NOTE: This interface is much less performant than the stream / batch interfaces for many tasks. Only use this interface for development / a small number of tasks.

func DoOrdered 

func DoOrdered(ctx context.Context, doer Doer, inputs chan *types.Any, parallelism int, cb CollectFunc) error

DoOrdered processes tasks in parallel, but returns outputs in order via the provided callback cb.

func List 

func List(ctx context.Context, svc Service, req *taskapi.ListTaskRequest, send func(info *taskapi.TaskInfo) error) error

List implements the functionality for an arbitrary service's ListTask gRPC

Types

type Cache 

type Cache interface {
	Get(ctx context.Context, key string) (output *types.Any, _ error)
	Put(ctx context.Context, key string, output *types.Any) error
}

type Claim 

type Claim struct {
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

func (*Claim) Descriptor 

func (*Claim) Descriptor() ([]byte, []int)

func (*Claim) Marshal 

func (m *Claim) Marshal() (dAtA []byte, err error)

func (*Claim) MarshalLogObject 

func (x *Claim) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*Claim) MarshalTo 

func (m *Claim) MarshalTo(dAtA []byte) (int, error)

func (*Claim) MarshalToSizedBuffer 

func (m *Claim) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Claim) ProtoMessage 

func (*Claim) ProtoMessage()

func (*Claim) Reset 

func (m *Claim) Reset()

func (*Claim) Size 

func (m *Claim) Size() (n int)

func (*Claim) String 

func (m *Claim) String() string

func (*Claim) Unmarshal 

func (m *Claim) Unmarshal(dAtA []byte) error

func (*Claim) XXX_DiscardUnknown 

func (m *Claim) XXX_DiscardUnknown()

func (*Claim) XXX_Marshal 

func (m *Claim) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Claim) XXX_Merge 

func (m *Claim) XXX_Merge(src proto.Message)

func (*Claim) XXX_Size 

func (m *Claim) XXX_Size() int

func (*Claim) XXX_Unmarshal 

func (m *Claim) XXX_Unmarshal(b []byte) error

type CollectFunc 

type CollectFunc = func(index int64, output *types.Any, _ error) error

CollectFunc is the type of a function that is used for collecting the output of a stream / batch of tasks. Index is the index of a task with respect to the order in which the task was created in the stream / batch.

type Doer 

type Doer interface {
	// Do creates and returns the results of processing a stream of tasks provided
	// by the input channel. The client should close the input channel when all tasks have
	// been sent (it does not need to be closed if the context is canceled). For each
	// task, the collect function will be called with the results.
	Do(ctx context.Context, inputChan chan *types.Any, cb CollectFunc) error
}

Doer is a doer of tasks. Refer to the DoOne and DoBatch helper functions if a simpler interface is desired.

type Group 

type Group struct {
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

func (*Group) Descriptor 

func (*Group) Descriptor() ([]byte, []int)

func (*Group) Marshal 

func (m *Group) Marshal() (dAtA []byte, err error)

func (*Group) MarshalLogObject 

func (x *Group) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*Group) MarshalTo 

func (m *Group) MarshalTo(dAtA []byte) (int, error)

func (*Group) MarshalToSizedBuffer 

func (m *Group) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Group) ProtoMessage 

func (*Group) ProtoMessage()

func (*Group) Reset 

func (m *Group) Reset()

func (*Group) Size 

func (m *Group) Size() (n int)

func (*Group) String 

func (m *Group) String() string

func (*Group) Unmarshal 

func (m *Group) Unmarshal(dAtA []byte) error

func (*Group) XXX_DiscardUnknown 

func (m *Group) XXX_DiscardUnknown()

func (*Group) XXX_Marshal 

func (m *Group) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Group) XXX_Merge 

func (m *Group) XXX_Merge(src proto.Message)

func (*Group) XXX_Size 

func (m *Group) XXX_Size() int

func (*Group) XXX_Unmarshal 

func (m *Group) XXX_Unmarshal(b []byte) error

type ProcessFunc 

type ProcessFunc = func(ctx context.Context, input *types.Any) (output *types.Any, _ error)

ProcessFunc is the type of a function that is use for processing a task. If an error occurs, then it should be returned. This error will be propagated back to the Doer that created the task.

type Service 

type Service interface {
	// NewDoer creates a Doer with the provided namespace and group.
	NewDoer(namespace, group string, cache Cache) Doer
	// NewSource creates a Source with the provided namespace.
	NewSource(namespace string) Source
	// List calls a function on every task under a namespace and group
	List(ctx context.Context, namespace, group string, cb func(namespace, group string, data *Task, claimed bool) error) error
}

Scheduling: A task managed by a Service has a group. The group is used for scheduling purposes. Scheduling is based on maximizing fairness for the groups, and the schedulable unit is a task.

func NewEtcdService 

func NewEtcdService(etcdClient *etcd.Client, etcdPrefix string) Service

type Source 

type Source interface {
	// Iterate iterates through tasks until the provided context is canceled.
	// For each task, the process function will be called and the results
	// will be returned to the Doer that created the task.
	Iterate(ctx context.Context, cb ProcessFunc) error
}

Source is a source of tasks.

type State 

type State int32
const (
	State_STATE_UNKNOWN State = 0
	State_RUNNING       State = 1
	State_SUCCESS       State = 2
	State_FAILURE       State = 3
)

func (State) EnumDescriptor 

func (State) EnumDescriptor() ([]byte, []int)

func (State) String 

func (x State) String() string

type Task 

type Task struct {
	ID                   string     `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
	State                State      `protobuf:"varint,2,opt,name=state,proto3,enum=task.State" json:"state,omitempty"`
	Input                *types.Any `protobuf:"bytes,3,opt,name=input,proto3" json:"input,omitempty"`
	Output               *types.Any `protobuf:"bytes,4,opt,name=output,proto3" json:"output,omitempty"`
	Reason               string     `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"`
	Index                int64      `protobuf:"varint,6,opt,name=index,proto3" json:"index,omitempty"`
	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
	XXX_unrecognized     []byte     `json:"-"`
	XXX_sizecache        int32      `json:"-"`
}

TODO: Consider splitting this up into separate structures for each state in a oneof.

func (*Task) Descriptor 

func (*Task) Descriptor() ([]byte, []int)

func (*Task) GetID 

func (m *Task) GetID() string

func (*Task) GetIndex 

func (m *Task) GetIndex() int64

func (*Task) GetInput 

func (m *Task) GetInput() *types.Any

func (*Task) GetOutput 

func (m *Task) GetOutput() *types.Any

func (*Task) GetReason 

func (m *Task) GetReason() string

func (*Task) GetState 

func (m *Task) GetState() State

func (*Task) Marshal 

func (m *Task) Marshal() (dAtA []byte, err error)

func (*Task) MarshalLogObject 

func (x *Task) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*Task) MarshalTo 

func (m *Task) MarshalTo(dAtA []byte) (int, error)

func (*Task) MarshalToSizedBuffer 

func (m *Task) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Task) ProtoMessage 

func (*Task) ProtoMessage()

func (*Task) Reset 

func (m *Task) Reset()

func (*Task) Size 

func (m *Task) Size() (n int)

func (*Task) String 

func (m *Task) String() string

func (*Task) Unmarshal 

func (m *Task) Unmarshal(dAtA []byte) error

func (*Task) XXX_DiscardUnknown 

func (m *Task) XXX_DiscardUnknown()

func (*Task) XXX_Marshal 

func (m *Task) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Task) XXX_Merge 

func (m *Task) XXX_Merge(src proto.Message)

func (*Task) XXX_Size 

func (m *Task) XXX_Size() int

func (*Task) XXX_Unmarshal 

func (m *Task) XXX_Unmarshal(b []byte) error

type TestTask 

type TestTask struct {
	ID                   string   `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

func (*TestTask) Descriptor 

func (*TestTask) Descriptor() ([]byte, []int)

func (*TestTask) GetID 

func (m *TestTask) GetID() string

func (*TestTask) Marshal 

func (m *TestTask) Marshal() (dAtA []byte, err error)

func (*TestTask) MarshalLogObject 

func (x *TestTask) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*TestTask) MarshalTo 

func (m *TestTask) MarshalTo(dAtA []byte) (int, error)

func (*TestTask) MarshalToSizedBuffer 

func (m *TestTask) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*TestTask) ProtoMessage 

func (*TestTask) ProtoMessage()

func (*TestTask) Reset 

func (m *TestTask) Reset()

func (*TestTask) Size 

func (m *TestTask) Size() (n int)

func (*TestTask) String 

func (m *TestTask) String() string

func (*TestTask) Unmarshal 

func (m *TestTask) Unmarshal(dAtA []byte) error

func (*TestTask) XXX_DiscardUnknown 

func (m *TestTask) XXX_DiscardUnknown()

func (*TestTask) XXX_Marshal 

func (m *TestTask) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*TestTask) XXX_Merge 

func (m *TestTask) XXX_Merge(src proto.Message)

func (*TestTask) XXX_Size 

func (m *TestTask) XXX_Size() int

func (*TestTask) XXX_Unmarshal 

func (m *TestTask) XXX_Unmarshal(b []byte) error

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.