stream_manager

Documentation

Index

• Constants
• func FindNodeIndex(extentInfo *pb.ExtentInfo, nodeID uint64) int
• func FormatRecoveryTaskName(extentID uint64) string
• type AllocExtentPolicy
• type DiskStatus
  • func (ds *DiskStatus) Free() uint64
  • func (ds *DiskStatus) Total() uint64
• type NodeStatus
  • func (ns *NodeStatus) Dead() bool
  • func (ns *NodeStatus) Free() uint64
  • func (ns *NodeStatus) GetConn() *grpc.ClientConn
  • func (ns *NodeStatus) IsHealthy() bool
  • func (ns *NodeStatus) LastEcho() time.Time
  • func (ns *NodeStatus) SetDead()
  • func (ns *NodeStatus) SetFree(f uint64)
  • func (ns *NodeStatus) SetTotal(t uint64)
  • func (ns *NodeStatus) Total() uint64
• type RecoveryTask
  • func (rt *RecoveryTask) RunningNode() uint64
  • func (rt *RecoveryTask) SetRunningNode(x uint64)
  • func (rt *RecoveryTask) SetStartTime()
  • func (rt *RecoveryTask) StartTime() time.Time
• type SimplePolicy
  • func (sp *SimplePolicy) AllocExtent(ns []*NodeStatus, count int, keepNodes []uint64) ([]*NodeStatus, error)
• type StreamManager
  • func NewStreamManager(etcd *embed.Etcd, client *clientv3.Client, config *manager.Config) *StreamManager
  • func (sm *StreamManager) AmLeader() bool
  • func (sm *StreamManager) CheckCommitLength(ctx context.Context, req *pb.CheckCommitLengthRequest) (*pb.CheckCommitLengthResponse, error)
  • func (sm *StreamManager) Close()
  • func (sm *StreamManager) CreateStream(ctx context.Context, req *pb.CreateStreamRequest) (*pb.CreateStreamResponse, error)
  • func (sm *StreamManager) ExtentInfo(ctx context.Context, req *pb.ExtentInfoRequest) (*pb.ExtentInfoResponse, error)
  • func (sm *StreamManager) LeaderLoop()
  • func (sm *StreamManager) MultiModifySplit(ctx context.Context, req *pb.MultiModifySplitRequest) (*pb.MultiModifySplitResponse, error)
  • func (sm *StreamManager) NodesInfo(ctx context.Context, req *pb.NodesInfoRequest) (*pb.NodesInfoResponse, error)
  • func (sm *StreamManager) RegisterGRPC(grpcServer *grpc.Server)
  • func (sm *StreamManager) RegisterNode(ctx context.Context, req *pb.RegisterNodeRequest) (*pb.RegisterNodeResponse, error)
  • func (sm *StreamManager) Status(ctx context.Context, req *pb.StatusRequest) (*pb.StatusResponse, error)
  • func (sm *StreamManager) StreamAllocExtent(ctx context.Context, req *pb.StreamAllocExtentRequest) (*pb.StreamAllocExtentResponse, error)
  • func (sm *StreamManager) StreamInfo(ctx context.Context, req *pb.StreamInfoRequest) (*pb.StreamInfoResponse, error)
  • func (sm *StreamManager) StreamPunchHoles(ctx context.Context, req *pb.PunchHolesRequest) (*pb.PunchHolesResponse, error)
  • func (sm *StreamManager) Truncate(ctx context.Context, req *pb.TruncateRequest) (*pb.TruncateResponse, error)
• type TaskPool
  • func NewTaskPool() *TaskPool
  • func (tp *TaskPool) GetFromExtent(extentID uint64) *pb.RecoveryTask
  • func (tp *TaskPool) GetFromNode(nodeID uint64) []*pb.RecoveryTask
  • func (tp *TaskPool) HasTask(extentID uint64) bool
  • func (tp *TaskPool) Insert(t *pb.RecoveryTask)
  • func (tp *TaskPool) Remove(extentID uint64)

Constants

const (
	IdKey = "AutumnSMIDKey"
)

Functions

func FindNodeIndex

func FindNodeIndex(extentInfo *pb.ExtentInfo, nodeID uint64) int

func FormatRecoveryTaskName

func FormatRecoveryTaskName(extentID uint64) string

Types

type AllocExtentPolicy

type AllocExtentPolicy interface {
	AllocExtent([]*NodeStatus, int, []uint64) ([]*NodeStatus, error)
}

type DiskStatus

type DiskStatus struct {
	pb.DiskInfo
	// contains filtered or unexported fields
}

func (*DiskStatus) Free

func (ds *DiskStatus) Free() uint64

func (*DiskStatus) Total

func (ds *DiskStatus) Total() uint64

type NodeStatus

type NodeStatus struct {
	pb.NodeInfo
	// contains filtered or unexported fields
}

func (*NodeStatus) Dead

func (ns *NodeStatus) Dead() bool

FIXME: could atomic.LoadPointer make it concise?

func (*NodeStatus) Free

func (ns *NodeStatus) Free() uint64

func (*NodeStatus) GetConn

func (ns *NodeStatus) GetConn() *grpc.ClientConn

func (*NodeStatus) IsHealthy

func (ns *NodeStatus) IsHealthy() bool

func (*NodeStatus) LastEcho

func (ns *NodeStatus) LastEcho() time.Time

func (*NodeStatus) SetDead

func (ns *NodeStatus) SetDead()

func (*NodeStatus) SetFree

func (ns *NodeStatus) SetFree(f uint64)

func (*NodeStatus) SetTotal

func (ns *NodeStatus) SetTotal(t uint64)

func (*NodeStatus) Total

func (ns *NodeStatus) Total() uint64

type RecoveryTask

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

func (*RecoveryTask) RunningNode

func (rt *RecoveryTask) RunningNode() uint64

func (*RecoveryTask) SetRunningNode

func (rt *RecoveryTask) SetRunningNode(x uint64)

func (*RecoveryTask) SetStartTime

func (rt *RecoveryTask) SetStartTime()

func (*RecoveryTask) StartTime

func (rt *RecoveryTask) StartTime() time.Time

type SimplePolicy

type SimplePolicy struct{}

func (*SimplePolicy) AllocExtent

func (sp *SimplePolicy) AllocExtent(ns []*NodeStatus, count int, keepNodes []uint64) ([]*NodeStatus, error)

type StreamManager

type StreamManager struct {
	ID uint64 //backend ETCD server's ID
	// contains filtered or unexported fields
}

func NewStreamManager

func NewStreamManager(etcd *embed.Etcd, client *clientv3.Client, config *manager.Config) *StreamManager

func (*StreamManager) AmLeader

func (sm *StreamManager) AmLeader() bool

func (*StreamManager) CheckCommitLength

func (sm *StreamManager) CheckCommitLength(ctx context.Context, req *pb.CheckCommitLengthRequest) (*pb.CheckCommitLengthResponse, error)

func (*StreamManager) Close

func (sm *StreamManager) Close()

func (*StreamManager) CreateStream

func (sm *StreamManager) CreateStream(ctx context.Context, req *pb.CreateStreamRequest) (*pb.CreateStreamResponse, error)

func (*StreamManager) ExtentInfo

func (sm *StreamManager) ExtentInfo(ctx context.Context, req *pb.ExtentInfoRequest) (*pb.ExtentInfoResponse, error)

func (*StreamManager) LeaderLoop

func (sm *StreamManager) LeaderLoop()

func (*StreamManager) MultiModifySplit

func (sm *StreamManager) MultiModifySplit(ctx context.Context, req *pb.MultiModifySplitRequest) (*pb.MultiModifySplitResponse, error)

func (*StreamManager) NodesInfo

func (sm *StreamManager) NodesInfo(ctx context.Context, req *pb.NodesInfoRequest) (*pb.NodesInfoResponse, error)

func (*StreamManager) RegisterGRPC

func (sm *StreamManager) RegisterGRPC(grpcServer *grpc.Server)

func (*StreamManager) RegisterNode

func (sm *StreamManager) RegisterNode(ctx context.Context, req *pb.RegisterNodeRequest) (*pb.RegisterNodeResponse, error)

func (*StreamManager) Status

func (sm *StreamManager) Status(ctx context.Context, req *pb.StatusRequest) (*pb.StatusResponse, error)

func (*StreamManager) StreamAllocExtent

func (sm *StreamManager) StreamAllocExtent(ctx context.Context, req *pb.StreamAllocExtentRequest) (*pb.StreamAllocExtentResponse, error)

func (*StreamManager) StreamInfo

func (sm *StreamManager) StreamInfo(ctx context.Context, req *pb.StreamInfoRequest) (*pb.StreamInfoResponse, error)

func (*StreamManager) StreamPunchHoles

func (sm *StreamManager) StreamPunchHoles(ctx context.Context, req *pb.PunchHolesRequest) (*pb.PunchHolesResponse, error)

func (*StreamManager) Truncate

func (sm *StreamManager) Truncate(ctx context.Context, req *pb.TruncateRequest) (*pb.TruncateResponse, error)

type TaskPool

type TaskPool struct {
	utils.SafeMutex
	// contains filtered or unexported fields
}

import "github.com/cornelk/hashmap" or use go-memdb as multiIndex

func NewTaskPool

func NewTaskPool() *TaskPool

func (*TaskPool) GetFromExtent

func (tp *TaskPool) GetFromExtent(extentID uint64) *pb.RecoveryTask

func (*TaskPool) GetFromNode

func (tp *TaskPool) GetFromNode(nodeID uint64) []*pb.RecoveryTask

func (*TaskPool) HasTask

func (tp *TaskPool) HasTask(extentID uint64) bool

func (*TaskPool) Insert

func (tp *TaskPool) Insert(t *pb.RecoveryTask)

func (*TaskPool) Remove

func (tp *TaskPool) Remove(extentID uint64)