worker

README

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

protoc -I worker worker/payload.proto --gofast_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

Package worker contains code for internal worker communication to perform queries and mutations.

Index

• Constants
• Variables
• func Connect(workerList []string, workerPort string)
• func GetOrAssignUidsOverNetwork(ctx context.Context, xidToUid map[string]uint64) (rerr error)
• func InitState(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 State
  • func (ws *State) PopulateShard(ctx context.Context, pred string, serverId int) error
• type WorkerClient
  • func NewWorkerClient(cc *grpc.ClientConn) WorkerClient
• type WorkerServer
• type Worker_PredicateDataClient
• type Worker_PredicateDataServer

Constants

const (
	// MB represents a megabyte.
	MB = 1 << 20
)

Variables

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

Functions

func Connect

func Connect(workerList []string, workerPort string)

Connect establishes a connection with other workers and sends the Hello rpc call to them.

func GetOrAssignUidsOverNetwork

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

GetOrAssignUidsOverNetwork gets or assigns uids corresponding to xids and writes them to the xidToUid map.

func InitState

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

InitState initializes the state on an instance with data,uid store and other meta.

func NewQuery

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

NewQuery creates a Query flatbuffer table, serializes and returns it.

func ProcessTaskOverNetwork

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

ProcessTaskOverNetwork is used to process the query and get the result from the instance which stores posting list corresponding to the predicate in the query.

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 slice 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{}

PayloadCodec is a custom codec that is that is used for internal worker communication.

func (*PayloadCodec) Marshal

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

Marshal marshals v into a Payload instance. v contains serialised data for a flatbuffer Query object.

func (*PayloadCodec) String

func (cb *PayloadCodec) String() string

func (*PayloadCodec) Unmarshal

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

Unmarshal unmarshals byte slice data into v.

type Pool

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

Pool is used to manage the grpc client connections for communicating with other worker instances.

func NewPool

func NewPool(addr string, maxCap int) *Pool

NewPool initializes an instance of Pool which is used to connect with other workers. The pool instance also has a buffered channel,conn with capacity maxCap that stores the connections.

func (*Pool) Get

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

Get returns a connection from the pool of connections or a new connection if the pool is empty.

func (*Pool) Put

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

Put returns a connection to the pool or closes and discards the connection incase the pool channel is at capacity.

type State

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

State stores the worker state.

func (*State) PopulateShard

func (ws *State) PopulateShard(ctx context.Context, pred string,
	serverId int) error

PopulateShard gets data for predicate pred from server with id serverId and writes it to RocksDB.

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)
	PredicateData(ctx context.Context, in *Payload, opts ...grpc.CallOption) (Worker_PredicateDataClient, 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)
	PredicateData(*Payload, Worker_PredicateDataServer) error
}

type Worker_PredicateDataClient

type Worker_PredicateDataClient interface {
	Recv() (*Payload, error)
	grpc.ClientStream
}

type Worker_PredicateDataServer

type Worker_PredicateDataServer interface {
	Send(*Payload) error
	grpc.ServerStream
}