diff --git a/weed-fs/src/pkg/directory/volume_mapping.go b/weed-fs/src/pkg/directory/volume_mapping.go index da41b3510..99aa90a5f 100644 --- a/weed-fs/src/pkg/directory/volume_mapping.go +++ b/weed-fs/src/pkg/directory/volume_mapping.go @@ -97,18 +97,9 @@ func (m *Mapper) remove(machine *Machine) { foundIndex = index } } - m.vid2machines[v.Id] = deleteFromSlice(foundIndex,m.vid2machines[v.Id]) + m.vid2machines[v.Id] = append(m.vid2machines[v.Id][:foundIndex], m.vid2machines[v.Id][foundIndex+1:]...) } } -func deleteFromSlice(i int, slice []*Machine) []*Machine{ - switch i { - case -1://do nothing - case 0: slice = slice[1:] - case len(slice)-1: slice = slice[:len(slice)-1] - default: slice = append(slice[:i], slice[i+1:]...) - } - return slice -} func (m *Mapper) StartRefreshWritableVolumes() { go func() { diff --git a/weed-fs/src/pkg/topology/data_node.go b/weed-fs/src/pkg/topology/data_node.go index 04c7cf111..2305dddd2 100644 --- a/weed-fs/src/pkg/topology/data_node.go +++ b/weed-fs/src/pkg/topology/data_node.go @@ -12,6 +12,7 @@ type DataNode struct { Port int PublicUrl string LastSeen int64 // unix time in seconds + Dead bool } func NewDataNode(id string) *DataNode { @@ -30,8 +31,8 @@ func (dn *DataNode) AddOrUpdateVolume(v *storage.VolumeInfo) { dn.volumes[v.Id] = v dn.UpAdjustActiveVolumeCountDelta(1) dn.UpAdjustMaxVolumeId(v.Id) - }else{ - dn.volumes[v.Id] = v + } else { + dn.volumes[v.Id] = v } } func (dn *DataNode) GetTopology() *Topology { diff --git a/weed-fs/src/pkg/topology/node.go b/weed-fs/src/pkg/topology/node.go index ddaf9f0b2..52315ca2f 100644 --- a/weed-fs/src/pkg/topology/node.go +++ b/weed-fs/src/pkg/topology/node.go @@ -20,7 +20,7 @@ type Node interface { setParent(Node) LinkChildNode(node Node) UnlinkChildNode(nodeId NodeId) - CollectWritableVolumes(freshThreshHold int64, volumeSizeLimit uint64) []storage.VolumeId + CollectDeadNodeAndFullVolumes(freshThreshHold int64, volumeSizeLimit uint64) IsDataNode() bool Children() map[NodeId]Node @@ -146,25 +146,34 @@ func (n *NodeImpl) UnlinkChildNode(nodeId NodeId) { } } -func (n *NodeImpl) CollectWritableVolumes(freshThreshHold int64, volumeSizeLimit uint64) []storage.VolumeId { - var ret []storage.VolumeId +func (n *NodeImpl) CollectDeadNodeAndFullVolumes(freshThreshHold int64, volumeSizeLimit uint64) { if n.IsRack() { for _, c := range n.Children() { dn := c.(*DataNode) //can not cast n to DataNode if dn.LastSeen > freshThreshHold { - continue + if !dn.Dead { + dn.Dead = true + n.GetTopology().chanDeadDataNodes <- dn + } } for _, v := range dn.volumes { if uint64(v.Size) < volumeSizeLimit { - ret = append(ret, v.Id) + n.GetTopology().chanFullVolumes <- v } } } } else { for _, c := range n.Children() { - ret = append(ret, c.CollectWritableVolumes(freshThreshHold, volumeSizeLimit)...) + c.CollectDeadNodeAndFullVolumes(freshThreshHold, volumeSizeLimit) } } - - return ret +} + +func (n *NodeImpl) GetTopology() *Topology{ + var p Node + p = n + for p.Parent() != nil { + p = p.Parent() + } + return p.(*Topology) } diff --git a/weed-fs/src/pkg/topology/rack.go b/weed-fs/src/pkg/topology/rack.go index c819feb00..bbcd594a2 100644 --- a/weed-fs/src/pkg/topology/rack.go +++ b/weed-fs/src/pkg/topology/rack.go @@ -30,6 +30,10 @@ func (r *Rack) GetOrCreateDataNode(ip string, port int, publicUrl string, maxVol dn := c.(*DataNode) if dn.MatchLocation(ip, port) { dn.LastSeen = time.Now().Unix() + if dn.Dead { + dn.Dead = false + r.GetTopology().chanRecoveredDataNodes <- dn + } dn.UpAdjustMaxVolumeCountDelta(maxVolumeCount - dn.maxVolumeCount) return dn } diff --git a/weed-fs/src/pkg/topology/topology.go b/weed-fs/src/pkg/topology/topology.go index a60768f14..3943f9555 100644 --- a/weed-fs/src/pkg/topology/topology.go +++ b/weed-fs/src/pkg/topology/topology.go @@ -20,6 +20,12 @@ type Topology struct { volumeSizeLimit uint64 sequence sequence.Sequencer + + chanDeadDataNodes chan *DataNode + chanRecoveredDataNodes chan *DataNode + chanFullVolumes chan *storage.VolumeInfo + chanIncomplemteVolumes chan *storage.VolumeInfo + chanRecoveredVolumes chan *storage.VolumeInfo } func NewTopology(id string, dirname string, filename string, volumeSizeLimit uint64, pulse int) *Topology { @@ -31,6 +37,11 @@ func NewTopology(id string, dirname string, filename string, volumeSizeLimit uin t.pulse = int64(pulse) t.volumeSizeLimit = volumeSizeLimit t.sequence = sequence.NewSequencer(dirname, filename) + t.chanDeadDataNodes = make(chan *DataNode) + t.chanRecoveredDataNodes = make(chan *DataNode) + t.chanFullVolumes = make(chan *storage.VolumeInfo) + t.chanIncomplemteVolumes = make(chan *storage.VolumeInfo) + t.chanRecoveredVolumes = make(chan *storage.VolumeInfo) return t } @@ -95,6 +106,12 @@ func (t *Topology) RegisterVolumes(volumeInfos []storage.VolumeInfo, ip string, t.RegisterVolumeLayout(&v, dn) } } +func (t *Topology) SetVolumeReadOnly(volumeInfo *storage.VolumeInfo) { + //TODO +} +func (t *Topology) UnRegisterDataNode(dn *DataNode) { + //TODO +} func (t *Topology) GetOrCreateDataCenter(ip string) *DataCenter { for _, c := range t.Children() { @@ -128,17 +145,24 @@ func (t *Topology) ToMap() interface{} { } func (t *Topology) StartRefreshWritableVolumes() { - go func() { - for { - t.refreshWritableVolumes() - time.Sleep(time.Duration(float32(t.pulse*1e3)*(1+rand.Float32())) * time.Millisecond) - } - }() -} - -func (t *Topology) refreshWritableVolumes() { - freshThreshHold := time.Now().Unix() - 3*t.pulse //5 times of sleep interval - //setting Writers, copy-on-write because of possible updating, this needs some future work! - t.CollectWritableVolumes(freshThreshHold, t.volumeSizeLimit) - //TODO: collect writable columes for each replication type + go func() { + for { + freshThreshHold := time.Now().Unix() - 3*t.pulse //5 times of sleep interval + t.CollectDeadNodeAndFullVolumes(freshThreshHold, t.volumeSizeLimit) + time.Sleep(time.Duration(float32(t.pulse*1e3)*(1+rand.Float32())) * time.Millisecond) + } + }() + go func() { + for { + select { + case <-t.chanIncomplemteVolumes: + case <-t.chanRecoveredVolumes: + case fv := <-t.chanFullVolumes: + t.SetVolumeReadOnly(fv) + case <-t.chanRecoveredDataNodes: + case dn := <-t.chanDeadDataNodes: + t.UnRegisterDataNode(dn) + } + } + }() }