worker

README

To update the protocol buffer definitions, run this from one directory above:

protoc -I worker worker/payload.proto --go_out=plugins=grpc:worker

Documentation

Overview

Package worker is a generated protocol buffer package.

It is generated from these files:

payload.proto

It has these top-level messages:

Payload

Index

• Variables
• func Connect(workerList []string)
• func GetOrAssignUidsOverNetwork(ctx context.Context, xidToUid *map[string]uint64) (rerr error)
• func Init(ps, uStore *store.Store, idx, numInst uint64)
• func NewQuery(attr string, uids []uint64) []byte
• func ProcessTaskOverNetwork(ctx context.Context, qu []byte) (result []byte, rerr error)
• func RegisterWorkerServer(s *grpc.Server, srv WorkerServer)
• type Mutations
  • func MutateOverNetwork(ctx context.Context, m Mutations) (left Mutations, rerr error)
  • func (m *Mutations) Decode(data []byte) error
  • func (m *Mutations) Encode() (data []byte, rerr error)
• type Payload
  • func (*Payload) Descriptor() ([]byte, []int)
  • func (m *Payload) Marshal() (data []byte, err error)
  • func (m *Payload) MarshalTo(data []byte) (int, error)
  • func (*Payload) ProtoMessage()
  • func (m *Payload) Reset()
  • func (m *Payload) Size() (n int)
  • func (m *Payload) String() string
  • func (m *Payload) Unmarshal(data []byte) error
• type PayloadCodec
  • func (cb *PayloadCodec) Marshal(v interface{}) ([]byte, error)
  • func (cb *PayloadCodec) String() string
  • func (cb *PayloadCodec) Unmarshal(data []byte, v interface{}) error
• type Pool
  • func NewPool(addr string, maxCap int) *Pool
  • func (p *Pool) Get() (*grpc.ClientConn, error)
  • func (p *Pool) Put(conn *grpc.ClientConn) error
• type WorkerClient
  • func NewWorkerClient(cc *grpc.ClientConn) WorkerClient
• type WorkerServer

Variables

var (
	ErrInvalidLengthPayload = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowPayload   = fmt.Errorf("proto: integer overflow")
)

Functions

func Connect

func Connect(workerList []string)

func GetOrAssignUidsOverNetwork

func GetOrAssignUidsOverNetwork(ctx context.Context, xidToUid *map[string]uint64) (rerr error)

func Init

func Init(ps, uStore *store.Store, idx, numInst uint64)

func NewQuery

func NewQuery(attr string, uids []uint64) []byte

func ProcessTaskOverNetwork

func ProcessTaskOverNetwork(ctx context.Context, qu []byte) (result []byte, rerr error)

func RegisterWorkerServer

func RegisterWorkerServer(s *grpc.Server, srv WorkerServer)

Types

type Mutations

type Mutations struct {
	Set []x.DirectedEdge
	Del []x.DirectedEdge
}

Mutations stores the directed edges for both the set and delete operations.

func MutateOverNetwork

func MutateOverNetwork(ctx context.Context, m Mutations) (left Mutations, rerr error)

MutateOverNetwork checks which instance should be running the mutations according to fingerprint of the predicate and sends it to that instance.

func (*Mutations) Decode

func (m *Mutations) Decode(data []byte) error

Decode decodes the mutation from a byte slice after receiving the byte slie over the network.

func (*Mutations) Encode

func (m *Mutations) Encode() (data []byte, rerr error)

Encode gob encodes the mutation which is then sent over to the instance which is supposed to run it.

type Payload

type Payload struct {
	Data []byte `protobuf:"bytes,1,opt,name=Data,json=data,proto3" json:"Data,omitempty"`
}

func (*Payload) Descriptor

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

func (*Payload) Marshal

func (m *Payload) Marshal() (data []byte, err error)

func (*Payload) MarshalTo

func (m *Payload) MarshalTo(data []byte) (int, error)

func (*Payload) ProtoMessage

func (*Payload) ProtoMessage()

func (*Payload) Reset

func (m *Payload) Reset()

func (*Payload) Size

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

func (*Payload) String

func (m *Payload) String() string

func (*Payload) Unmarshal

func (m *Payload) Unmarshal(data []byte) error

type PayloadCodec

type PayloadCodec struct{}

func (*PayloadCodec) Marshal

func (cb *PayloadCodec) Marshal(v interface{}) ([]byte, error)

func (*PayloadCodec) String

func (cb *PayloadCodec) String() string

func (*PayloadCodec) Unmarshal

func (cb *PayloadCodec) Unmarshal(data []byte, v interface{}) error

type Pool

type Pool struct {
	Addr string
	// contains filtered or unexported fields
}

func NewPool

func NewPool(addr string, maxCap int) *Pool

func (*Pool) Get

func (p *Pool) Get() (*grpc.ClientConn, error)

func (*Pool) Put

func (p *Pool) Put(conn *grpc.ClientConn) error

type WorkerClient

type WorkerClient interface {
	Hello(ctx context.Context, in *Payload, opts ...grpc.CallOption) (*Payload, error)
	GetOrAssign(ctx context.Context, in *Payload, opts ...grpc.CallOption) (*Payload, error)
	Mutate(ctx context.Context, in *Payload, opts ...grpc.CallOption) (*Payload, error)
	ServeTask(ctx context.Context, in *Payload, opts ...grpc.CallOption) (*Payload, error)
}

func NewWorkerClient

func NewWorkerClient(cc *grpc.ClientConn) WorkerClient

type WorkerServer

type WorkerServer interface {
	Hello(context.Context, *Payload) (*Payload, error)
	GetOrAssign(context.Context, *Payload) (*Payload, error)
	Mutate(context.Context, *Payload) (*Payload, error)
	ServeTask(context.Context, *Payload) (*Payload, error)
}