executor

README

Structure

server

executor-server is in charge of:

• maintain heartbeat with master.
• transfer the tasks from master to runtime engine.

runtime

The task runtime maintains some important structures and implements the interface:

• runtime, the singleton instance organizes and executes tasks.
  • Run function that actually drives the execution of the tasks and is thread-safe.
  • Queue field that contains the runnable tasks.
• taskContainer wraps the logic of operator and maintains the status and input/output channels.
  • Poll triggers a executing action of the task, which will return quickly. During the poll action, the task will
    • read inputs channels, if the channels are empty, return blocked. The read action fetches a batch of data.
  • status records the current status of task, including
    • runnable means the task is in the Queue and can run immediately.
    • blocked means the task is waiting someone to trigger, say output data consuming, input data arriving or i/o task completed.
    • awaiting means this task has been awoken by someone. If the Poll function ends and checks this status, we should put this task back to queue and reset the status to runnable.
• Operator actually implement the user logic. Every operator is contained by a task. Ideally, operators can construct an operating tree like a typical sql engine.
  • prepare is called when constructing the operator and prepare some resources.
  • close releases all resources gracefully.
  • next(ctx *taskContext, r *record, index int) accepts the incoming data and returns the result data or blocked status. index indicates which input the record comes from.
  • NextWantedInputIdx returns a integer indicating which input the task should read in this poll. In normal cases it returns non-negative value. There are two special value:
    • DontNeedData suggests not to read data from input. It happens when the operator reads data from grpc streams or the operator was blocked last time and need to digest current blocking record at this time.
    • DontRequireData. Given a union operator, it can read from any inputs without caring about the read order. In this case, we return this status.

Documentation

Index

• type Config
  • func GetDefaultExecutorConfig() *Config
  • func (c *Config) Adjust() (err error)
  • func (c *Config) String() string
  • func (c *Config) Toml() (string, error)
• type Server
  • func NewServer(cfg *Config) *Server
  • func (s *Server) ConfirmDispatchTask(ctx context.Context, req *pb.ConfirmDispatchTaskRequest) (*pb.ConfirmDispatchTaskResponse, error)
  • func (s *Server) PreDispatchTask(ctx context.Context, req *pb.PreDispatchTaskRequest) (*pb.PreDispatchTaskResponse, error)
  • func (s *Server) Run(ctx context.Context) error
  • func (s *Server) Stop()

Types

type Config

type Config struct {
	Name string `toml:"name" json:"name"`

	LogConf logutil.Config `toml:"log" json:"log"`

	Join          string `toml:"join" json:"join" `
	Addr          string `toml:"addr" json:"addr"`
	AdvertiseAddr string `toml:"advertise-addr" json:"advertise-addr"`

	Labels map[string]string `toml:"labels" json:"labels"`
	// EnableGCTuning enables a GC tuning mechanism that adjusts the GC frequency
	// according to the used memory with reference to the total memory. It can be
	// enabled when the executor can consume almost all the memory of the
	// container/machine.
	EnableGCTuning bool `toml:"enable-gc-tuning" json:"enable-gc-tuning"`

	// TODO: in the future executors should share a same ttl from server-master
	KeepAliveTTLStr      string `toml:"keepalive-ttl" json:"keepalive-ttl"`
	KeepAliveIntervalStr string `toml:"keepalive-interval" json:"keepalive-interval"`
	RPCTimeoutStr        string `toml:"rpc-timeout" json:"rpc-timeout"`

	KeepAliveTTL      time.Duration `toml:"-" json:"-"`
	KeepAliveInterval time.Duration `toml:"-" json:"-"`
	RPCTimeout        time.Duration `toml:"-" json:"-"`

	Security *security.Credential `toml:"security" json:"security"`
}

Config is the configuration.

func GetDefaultExecutorConfig

func GetDefaultExecutorConfig() *Config

GetDefaultExecutorConfig returns a default executor config

func (*Config) Adjust

func (c *Config) Adjust() (err error)

Adjust adjusts the executor configuration

func (*Config) String

func (c *Config) String() string

String implements fmt.Stringer

func (*Config) Toml

func (c *Config) Toml() (string, error)

Toml returns TOML format representation of config.

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server is an executor server.

func NewServer

func NewServer(cfg *Config) *Server

NewServer creates a new executor server instance

func (*Server) ConfirmDispatchTask

func (s *Server) ConfirmDispatchTask(ctx context.Context, req *pb.ConfirmDispatchTaskRequest) (*pb.ConfirmDispatchTaskResponse, error)

ConfirmDispatchTask implements Executor.ConfirmDispatchTask

func (*Server) PreDispatchTask

func (s *Server) PreDispatchTask(ctx context.Context, req *pb.PreDispatchTaskRequest) (*pb.PreDispatchTaskResponse, error)

PreDispatchTask implements Executor.PreDispatchTask

func (*Server) Run

func (s *Server) Run(ctx context.Context) error

Run drives server logic in independent background goroutines, and use error group to collect errors.

func (*Server) Stop

func (s *Server) Stop()

Stop stops all running goroutines and releases resources in Server