clusterio

Documentation

Index

• Variables
• type Agent
  • func NewAgent(nodeClient NodeClient, partitionResolver PartitionResolver) *Agent
  • func (agent *Agent) Batch(ctx context.Context, siteID string, bucket string, updateBatch *UpdateBatch) (int, int, error)
  • func (agent *Agent) CancelAll()
  • func (agent *Agent) Get(ctx context.Context, siteID string, bucket string, keys [][]byte) ([]*SiblingSet, error)
  • func (agent *Agent) GetMatches(ctx context.Context, siteID string, bucket string, keys [][]byte) (SiblingSetIterator, error)
  • func (agent *Agent) Merge(ctx context.Context, siteID string, bucket string, ...) (int, int, error)
  • func (agent *Agent) NQuorum(replicas int) int
  • func (agent *Agent) RelayStatus(ctx context.Context, siteID string, relayID string) (RelayStatus, error)
• type ClusterIOAgent
• type NodeClient
• type NodeReadMerger
• type NodeReadRepairer
• type PartitionResolver
• type ReadMerger
  • func NewReadMerger(bucket string) *ReadMerger
  • func (readMerger *ReadMerger) Get(key string) *SiblingSet
  • func (readMerger *ReadMerger) InsertKeyReplica(nodeID uint64, key string, siblingSet *SiblingSet)
  • func (readMerger *ReadMerger) Nodes() map[uint64]bool
  • func (readMerger *ReadMerger) Patch(nodeID uint64) map[string]*SiblingSet
• type ReadRepairer
  • func NewReadRepairer(nodeClient NodeClient) *ReadRepairer
  • func (readRepairer *ReadRepairer) BeginRepair(partition uint64, siteID string, bucket string, readMerger NodeReadMerger)
  • func (readRepairer *ReadRepairer) StopRepairs()
• type SiblingSetMergeIterator
  • func NewSiblingSetMergeIterator(readMerger NodeReadMerger) *SiblingSetMergeIterator
  • func (iter *SiblingSetMergeIterator) AddKey(prefix string, key string)
  • func (iter *SiblingSetMergeIterator) Error() error
  • func (iter *SiblingSetMergeIterator) Key() []byte
  • func (iter *SiblingSetMergeIterator) LocalVersion() uint64
  • func (iter *SiblingSetMergeIterator) Next() bool
  • func (iter *SiblingSetMergeIterator) Prefix() []byte
  • func (iter *SiblingSetMergeIterator) Release()
  • func (iter *SiblingSetMergeIterator) SortKeys()
  • func (iter *SiblingSetMergeIterator) Value() *SiblingSet

Variables

var DefaultTimeout time.Duration = time.Second * 20

Types

type Agent

type Agent struct {
	PartitionResolver PartitionResolver
	NodeClient        NodeClient
	NodeReadRepairer  NodeReadRepairer
	Timeout           time.Duration
	// contains filtered or unexported fields
}

func NewAgent

func NewAgent(nodeClient NodeClient, partitionResolver PartitionResolver) *Agent

func (*Agent) Batch

func (agent *Agent) Batch(ctx context.Context, siteID string, bucket string, updateBatch *UpdateBatch) (int, int, error)

func (*Agent) CancelAll

func (agent *Agent) CancelAll()

func (*Agent) Get

func (agent *Agent) Get(ctx context.Context, siteID string, bucket string, keys [][]byte) ([]*SiblingSet, error)

func (*Agent) GetMatches

func (agent *Agent) GetMatches(ctx context.Context, siteID string, bucket string, keys [][]byte) (SiblingSetIterator, error)

func (*Agent) Merge

func (agent *Agent) Merge(ctx context.Context, siteID string, bucket string, patch map[string]*SiblingSet) (int, int, error)

func (*Agent) NQuorum

func (agent *Agent) NQuorum(replicas int) int

func (*Agent) RelayStatus

func (agent *Agent) RelayStatus(ctx context.Context, siteID string, relayID string) (RelayStatus, error)

type ClusterIOAgent

type ClusterIOAgent interface {
	Merge(ctx context.Context, siteID string, bucket string, patch map[string]*SiblingSet) (replicas int, nApplied int, err error)
	Batch(ctx context.Context, siteID string, bucket string, updateBatch *UpdateBatch) (replicas int, nApplied int, err error)
	Get(ctx context.Context, siteID string, bucket string, keys [][]byte) ([]*SiblingSet, error)
	GetMatches(ctx context.Context, siteID string, bucket string, keys [][]byte) (SiblingSetIterator, error)
	RelayStatus(ctx context.Context, siteID string, relayID string) (RelayStatus, error)
	CancelAll()
}

type NodeClient

type NodeClient interface {
	Merge(ctx context.Context, nodeID uint64, partition uint64, siteID string, bucket string, patch map[string]*SiblingSet, broadcastToRelays bool) error
	Batch(ctx context.Context, nodeID uint64, partition uint64, siteID string, bucket string, updateBatch *UpdateBatch) (map[string]*SiblingSet, error)
	Get(ctx context.Context, nodeID uint64, partition uint64, siteID string, bucket string, keys [][]byte) ([]*SiblingSet, error)
	GetMatches(ctx context.Context, nodeID uint64, partition uint64, siteID string, bucket string, keys [][]byte) (SiblingSetIterator, error)
	RelayStatus(ctx context.Context, nodeID uint64, siteID string, relayID string) (RelayStatus, error)
	LocalNodeID() uint64
}

type NodeReadMerger

type NodeReadMerger interface {
	// Add to the pool of replicas for this key
	InsertKeyReplica(nodeID uint64, key string, siblingSet *SiblingSet)
	// Get the merged set for this key
	Get(key string) *SiblingSet
	// Obtain a patch that needs to be merged into the specified node to bring it up to date
	// for any keys for which there are updates that it has not received
	Patch(nodeID uint64) map[string]*SiblingSet
	// Get a set of nodes involved in the read merger
	Nodes() map[uint64]bool
}

type NodeReadRepairer

type NodeReadRepairer interface {
	BeginRepair(partition uint64, siteID string, bucket string, readMerger NodeReadMerger)
	StopRepairs()
}

type PartitionResolver

type PartitionResolver interface {
	Partition(partitioningKey string) uint64
	ReplicaNodes(partition uint64) []uint64
}

type ReadMerger

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

func NewReadMerger

func NewReadMerger(bucket string) *ReadMerger

func (*ReadMerger) Get

func (readMerger *ReadMerger) Get(key string) *SiblingSet

func (*ReadMerger) InsertKeyReplica

func (readMerger *ReadMerger) InsertKeyReplica(nodeID uint64, key string, siblingSet *SiblingSet)

func (*ReadMerger) Nodes

func (readMerger *ReadMerger) Nodes() map[uint64]bool

func (*ReadMerger) Patch

func (readMerger *ReadMerger) Patch(nodeID uint64) map[string]*SiblingSet

type ReadRepairer

type ReadRepairer struct {
	NodeClient NodeClient
	Timeout    time.Duration
	// contains filtered or unexported fields
}

func NewReadRepairer

func NewReadRepairer(nodeClient NodeClient) *ReadRepairer

func (*ReadRepairer) BeginRepair

func (readRepairer *ReadRepairer) BeginRepair(partition uint64, siteID string, bucket string, readMerger NodeReadMerger)

func (*ReadRepairer) StopRepairs

func (readRepairer *ReadRepairer) StopRepairs()

type SiblingSetMergeIterator

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

func NewSiblingSetMergeIterator

func NewSiblingSetMergeIterator(readMerger NodeReadMerger) *SiblingSetMergeIterator

func (*SiblingSetMergeIterator) AddKey

func (iter *SiblingSetMergeIterator) AddKey(prefix string, key string)

func (*SiblingSetMergeIterator) Error

func (iter *SiblingSetMergeIterator) Error() error

func (*SiblingSetMergeIterator) Key

func (iter *SiblingSetMergeIterator) Key() []byte

func (*SiblingSetMergeIterator) LocalVersion

func (iter *SiblingSetMergeIterator) LocalVersion() uint64

func (*SiblingSetMergeIterator) Next

func (iter *SiblingSetMergeIterator) Next() bool

func (*SiblingSetMergeIterator) Prefix

func (iter *SiblingSetMergeIterator) Prefix() []byte

func (*SiblingSetMergeIterator) Release

func (iter *SiblingSetMergeIterator) Release()

func (*SiblingSetMergeIterator) SortKeys

func (iter *SiblingSetMergeIterator) SortKeys()

func (*SiblingSetMergeIterator) Value

func (iter *SiblingSetMergeIterator) Value() *SiblingSet