executors

Documentation

Index

• func FromDelivery(d amqp.Delivery) *rabbidmqMessage
• type AsyncExecutor
  • func (p AsyncExecutor) Execute(job *falco.Job, submittable falco.Submittable, collector falco.ResultCollector) error
• type DEQueue
• type DEQueueMessage
• type DeadlineStragglerStrategy
  • func (d DeadlineStragglerStrategy) SelectStranglers(job *falco.Job, p DistributedExecutor) ([]falco.InvocationPayload, []falco.InvocationPayload, int)
• type DistributedExecutor
  • func (p DistributedExecutor) Execute(job *falco.Job, submittable falco.Submittable, writer falco.ResultCollector) error
• type MeanBackoffStragglerStrategy
  • func (m MeanBackoffStragglerStrategy) SelectStranglers(job *falco.Job, p DistributedExecutor) ([]falco.InvocationPayload, []falco.InvocationPayload, int)
• type MockMessage
  • func FromMeasurement(measurement falco.Measurement) MockMessage
  • func (m MockMessage) Body() []byte
• type MockQueue
  • func (m *MockQueue) Add(jobName string, measurement falco.Measurement)
  • func (m *MockQueue) Close() error
  • func (m *MockQueue) Connect() error
  • func (m *MockQueue) Consume(s string) (<-chan DEQueueMessage, error)
  • func (m *MockQueue) Delete(s string) error
  • func (m *MockQueue) Open(s string) error
  • func (m *MockQueue) Purge(s string) error
  • func (m *MockQueue) Setup(c *falco.Context) error
• type MockSubmittable
  • func NewMockSubmittable(t *testing.T, delay time.Duration, queue *MockQueue) *MockSubmittable
  • func (m MockSubmittable) Collect(job *falco.Job, i <-chan map[string]interface{}, ...) error
  • func (m MockSubmittable) Deploy(deployable falco.Deployable) (falco.Deployment, error)
  • func (m MockSubmittable) Invoke(deployment falco.Deployment, payload falco.InvocationPayload, ...) error
  • func (m MockSubmittable) Remove(deployment falco.Deployment) error
  • func (m MockSubmittable) Scale(deployment falco.Deployment, options ...falco.ScaleOptions) (falco.Deployment, error)
  • func (m MockSubmittable) Submit(job *falco.Job, payload falco.InvocationPayload, ...) error
• type ParallelExecutor
  • func (o ParallelExecutor) Execute(job *falco.Job, submittable falco.Submittable, collector falco.ResultCollector) error
• type RabbitMQWrapper
  • func (r RabbitMQWrapper) Close() error
  • func (r RabbitMQWrapper) Connect() error
  • func (r RabbitMQWrapper) Consume(name string) (<-chan DEQueueMessage, error)
  • func (r RabbitMQWrapper) Delete(name string) error
  • func (r RabbitMQWrapper) Open(name string) error
  • func (r RabbitMQWrapper) Purge(name string) error
  • func (r RabbitMQWrapper) Setup(c *falco.Context) error
• type SequentialExecutor
  • func (p SequentialExecutor) Execute(job *falco.Job, submittable falco.Submittable, collector falco.ResultCollector) error
• type StragglerStrategy

Functions

func FromDelivery

func FromDelivery(d amqp.Delivery) *rabbidmqMessage

Types

type AsyncExecutor

type AsyncExecutor struct {
	Timeout time.Duration
}

func (AsyncExecutor) Execute

func (p AsyncExecutor) Execute(job *falco.Job, submittable falco.Submittable, collector falco.ResultCollector) error

type DEQueue

type DEQueue interface {
	Setup(*falco.Context) error
	Connect() error
	Close() error
	Open(string) error
	Purge(string) error
	Delete(string) error
	Consume(string) (<-chan DEQueueMessage, error)
}

type DEQueueMessage

type DEQueueMessage interface {
	Body() []byte
}

type DeadlineStragglerStrategy

type DeadlineStragglerStrategy struct {
	DeadlineDuration time.Duration
	ReTryThreshold   int8
}

func (DeadlineStragglerStrategy) SelectStranglers

func (d DeadlineStragglerStrategy) SelectStranglers(job *falco.Job, p DistributedExecutor) ([]falco.InvocationPayload, []falco.InvocationPayload, int)

type DistributedExecutor

type DistributedExecutor struct {
	Queue        DEQueue
	Timeout      time.Duration
	TestInterval time.Duration
	Strategy     StragglerStrategy
}

func (DistributedExecutor) Execute

func (p DistributedExecutor) Execute(job *falco.Job, submittable falco.Submittable, writer falco.ResultCollector) error

type MeanBackoffStragglerStrategy

type MeanBackoffStragglerStrategy struct {
	//maximum number of re-submissions before a straggler is makred as a failed request
	ReTryThreshold int8
	//the percentage of total task needed to calculate the mean execution time used as a threashold
	MinimumSampleSize *float32
	//addtional time over the mean exection time that a task is still not marked as a straggler
	Graceperiod time.Duration
}

func (MeanBackoffStragglerStrategy) SelectStranglers

func (m MeanBackoffStragglerStrategy) SelectStranglers(job *falco.Job, p DistributedExecutor) ([]falco.InvocationPayload, []falco.InvocationPayload, int)

type MockMessage

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

func FromMeasurement

func FromMeasurement(measurement falco.Measurement) MockMessage

func (MockMessage) Body

func (m MockMessage) Body() []byte

type MockQueue

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

func (*MockQueue) Add

func (m *MockQueue) Add(jobName string, measurement falco.Measurement)

func (*MockQueue) Close

func (m *MockQueue) Close() error

func (*MockQueue) Connect

func (m *MockQueue) Connect() error

func (*MockQueue) Consume

func (m *MockQueue) Consume(s string) (<-chan DEQueueMessage, error)

func (*MockQueue) Delete

func (m *MockQueue) Delete(s string) error

func (*MockQueue) Open

func (m *MockQueue) Open(s string) error

func (*MockQueue) Purge

func (m *MockQueue) Purge(s string) error

func (*MockQueue) Setup

func (m *MockQueue) Setup(c *falco.Context) error

type MockSubmittable

type MockSubmittable struct {
	Submitted   *atomic.Int32
	Resubmitted *atomic.Int32
	// contains filtered or unexported fields
}

func NewMockSubmittable

func NewMockSubmittable(t *testing.T, delay time.Duration, queue *MockQueue) *MockSubmittable

func (MockSubmittable) Collect

func (m MockSubmittable) Collect(job *falco.Job, i <-chan map[string]interface{}, collector falco.ResultCollector, options ...falco.InvocableOptions) error

func (MockSubmittable) Deploy

func (m MockSubmittable) Deploy(deployable falco.Deployable) (falco.Deployment, error)

func (MockSubmittable) Invoke

func (m MockSubmittable) Invoke(deployment falco.Deployment, payload falco.InvocationPayload, collector falco.ResultCollector) error

func (MockSubmittable) Remove

func (m MockSubmittable) Remove(deployment falco.Deployment) error

func (MockSubmittable) Scale

func (m MockSubmittable) Scale(deployment falco.Deployment, options ...falco.ScaleOptions) (falco.Deployment, error)

func (MockSubmittable) Submit

func (m MockSubmittable) Submit(job *falco.Job, payload falco.InvocationPayload, c chan<- map[string]interface{}, options ...falco.InvocableOptions) error

type ParallelExecutor

type ParallelExecutor struct {
	Threads int
}

func (ParallelExecutor) Execute

func (o ParallelExecutor) Execute(job *falco.Job, submittable falco.Submittable, collector falco.ResultCollector) error

type RabbitMQWrapper

type RabbitMQWrapper struct {
	QueueConnection *amqp.Connection
	Channel         *amqp.Channel
	Queues          map[string]amqp.Queue
}

func (RabbitMQWrapper) Close

func (r RabbitMQWrapper) Close() error

func (RabbitMQWrapper) Connect

func (r RabbitMQWrapper) Connect() error

func (RabbitMQWrapper) Consume

func (r RabbitMQWrapper) Consume(name string) (<-chan DEQueueMessage, error)

func (RabbitMQWrapper) Delete

func (r RabbitMQWrapper) Delete(name string) error

func (RabbitMQWrapper) Open

func (r RabbitMQWrapper) Open(name string) error

func (RabbitMQWrapper) Purge

func (r RabbitMQWrapper) Purge(name string) error

func (RabbitMQWrapper) Setup

func (r RabbitMQWrapper) Setup(c *falco.Context) error

Setup the following values must be set in the context: rmquser - username of rabidmq (default guest) rmqpass - password of rabidmq user (default guest) rmq - address (ip or hostname) for rabidmq (default localhost) rmqport - port for rabidmq (default 5672)

type SequentialExecutor

type SequentialExecutor struct {
}

func (SequentialExecutor) Execute

func (p SequentialExecutor) Execute(job *falco.Job,
	submittable falco.Submittable,
	collector falco.ResultCollector) error

type StragglerStrategy

type StragglerStrategy interface {
	SelectStranglers(job *falco.Job, p DistributedExecutor) ([]falco.InvocationPayload, []falco.InvocationPayload, int)
}