server

Documentation

Index

• Variables
• type ClientBeat
• type ClientData
  • func (worker *ClientData) ConnectionCount() int
  • func (worker *ClientData) IsConsumer() bool
  • func (worker *ClientData) IsQuiet() bool
  • func (worker *ClientData) Signal(newstate WorkerState)
• type Connection
  • func (c *Connection) Close() error
  • func (c *Connection) Error(cmd string, err error) error
  • func (c *Connection) Number(val int) error
  • func (c *Connection) Ok() error
  • func (c *Connection) Result(msg []byte) error
• type RuntimeStats
• type Server
  • func NewServer(opts *ServerOptions) (*Server, error)
  • func (s *Server) AddTask(everySec int64, task Taskable)
  • func (s *Server) Boot() error
  • func (s *Server) CurrentState() (map[string]interface{}, error)
  • func (s *Server) Heartbeats() map[string]*ClientData
  • func (s *Server) Manager() manager.Manager
  • func (s *Server) Register(x Subsystem)
  • func (s *Server) Reload()
  • func (s *Server) Run() error
  • func (s *Server) Stop(f func())
  • func (s *Server) Stopper() chan bool
  • func (s *Server) Store() storage.Store
• type ServerOptions
  • func (so *ServerOptions) Config(subsys string, key string, defval interface{}) interface{}
  • func (so *ServerOptions) String(subsys string, key string, defval string) string
• type Subsystem
• type Taskable
• type WorkerState

Variables

var (
	AlwaysMatch = func(value string) bool {
		return true
	}
)

var CommandSet = map[string]command{
	"END":    end,
	"PUSH":   push,
	"PUSHB":  pushBulk,
	"FETCH":  fetch,
	"ACK":    ack,
	"FAIL":   fail,
	"BEAT":   heartbeat,
	"INFO":   info,
	"FLUSH":  flush,
	"MUTATE": mutate,
	"BATCH":  batch,
	"TRACK":  track,
	"QUEUE":  queue,
}

var DefaultMaxPoolSize = 1000

Types

type ClientBeat

type ClientBeat struct {
	CurrentState string `json:"current_state"`
	Wid          string `json:"wid"`
	RssKb        int64  `json:"rss_kb"`
}

type ClientData

type ClientData struct {
	Hostname     string   `json:"hostname"`
	Wid          string   `json:"wid"`
	Pid          int      `json:"pid"`
	RssKb        int64    `json:"rss_kb"`
	Labels       []string `json:"labels"`
	PasswordHash string   `json:"pwdhash"`
	Username     string   `json:"username"`
	Version      uint8    `json:"v"`
	StartedAt    time.Time
	// contains filtered or unexported fields
}

This represents a single client process. It may have many network connections open to Faktory.

A client can be a producer AND/OR consumer of jobs. Typically a process will either only produce jobs (like a webapp pushing jobs) or produce/consume jobs (like a faktory worker process where a job can create other jobs while executing).

Each Faktory worker process should send a BEAT command every 15 seconds. Only consumers should send a BEAT. If Faktory does not receive a BEAT from a worker process within 60 seconds, it expires and is removed from the Busy page.

From Faktory's POV, the worker can BEAT again and resume normal operations, e.g. due to a network partition. If a process dies, it will be removed after 1 minute and its jobs recovered after the job reservation timeout has passed (typically 30 minutes).

A worker process has a simple three-state lifecycle:

running -> quiet -> terminate

- Running means the worker is alive and processing jobs. - Quiet means the worker should stop FETCHing new jobs but continue working on existing jobs. It should not exit, even if no jobs are processing. - Terminate means the worker should exit within N seconds, where N is recommended to be 30 seconds. In practice, faktory_worker_ruby waits up to 25 seconds and any threads that are still busy are forcefully killed and their associated jobs reported as FAILed so they will be retried shortly.

A worker process should never stop sending BEAT. Even after "quiet" or "terminate", the BEAT should continue, only stopping due to process exit(). Workers should never move backward in state - you cannot "unquiet" a worker, it must be restarted.

Workers will typically also respond to standard Unix signals. faktory_worker_ruby uses TSTP ("Threads SToP") as the quiet signal and TERM as the terminate signal.

func (*ClientData) ConnectionCount

func (worker *ClientData) ConnectionCount() int

func (*ClientData) IsConsumer

func (worker *ClientData) IsConsumer() bool

func (*ClientData) IsQuiet

func (worker *ClientData) IsQuiet() bool

func (*ClientData) Signal

func (worker *ClientData) Signal(newstate WorkerState)

* Send "quiet" or "terminate" to the given client * worker process. Other signals are undefined.

type Connection

type Connection struct {
	context.Context
	// contains filtered or unexported fields
}

Represents a connection to a faktory client.

faktory reuses the same wire protocol as Redis: RESP. It's a nice trade-off between human-readable and efficient. Shout out to antirez for his nice design document on it. https://redis.io/topics/protocol

func (*Connection) Close

func (c *Connection) Close() error

func (*Connection) Error

func (c *Connection) Error(cmd string, err error) error

func (*Connection) Number

func (c *Connection) Number(val int) error

func (*Connection) Ok

func (c *Connection) Ok() error

func (*Connection) Result

func (c *Connection) Result(msg []byte) error

type RuntimeStats

type RuntimeStats struct {
	Connections uint64
	Commands    uint64
	StartedAt   time.Time
}

type Server

type Server struct {
	Options    *ServerOptions
	Stats      *RuntimeStats
	Subsystems []Subsystem
	// contains filtered or unexported fields
}

func NewServer

func NewServer(opts *ServerOptions) (*Server, error)

func (*Server) AddTask

func (s *Server) AddTask(everySec int64, task Taskable)

func (*Server) Boot

func (s *Server) Boot() error

func (*Server) CurrentState

func (s *Server) CurrentState() (map[string]interface{}, error)

func (*Server) Heartbeats

func (s *Server) Heartbeats() map[string]*ClientData

func (*Server) Manager

func (s *Server) Manager() manager.Manager

func (*Server) Register

func (s *Server) Register(x Subsystem)

register a global handler to be called when the Server instance has finished booting but before it starts listening.

func (*Server) Reload

func (s *Server) Reload()

func (*Server) Run

func (s *Server) Run() error

func (*Server) Stop

func (s *Server) Stop(f func())

func (*Server) Stopper

func (s *Server) Stopper() chan bool

func (*Server) Store

func (s *Server) Store() storage.Store

type ServerOptions

type ServerOptions struct {
	Binding          string
	StorageDirectory string
	RedisSock        string
	ConfigDirectory  string
	Environment      string
	Password         string
	PoolSize         int
	GlobalConfig     map[string]interface{}
}

func (*ServerOptions) Config

func (so *ServerOptions) Config(subsys string, key string, defval interface{}) interface{}

func (*ServerOptions) String

func (so *ServerOptions) String(subsys string, key string, defval string) string

type Subsystem

type Subsystem interface {
	Name() string

	// Called when the server is configured but before it starts accepting client connections.
	Start(*Server) error

	// Called every time Faktory reloads the global config for the Server.
	// Each subsystem is responsible for diffing its own config and making
	// necessary changes.
	Reload(*Server) error
}

type Taskable

type Taskable interface {
	Name() string
	Execute(context.Context) error
	Stats(context.Context) map[string]interface{}
}

type WorkerState

type WorkerState int

const (
	Running WorkerState = iota
	Quiet
	Terminate
)