seaweedfs/go/topology/topology_event_handling.go
Chris Lu 41143b3b78 toughen weedfs clustering, adding synchronizing max volume id among
peers in order to avoid the same volume id being assigned twice
1. moving raft.Server to topology
2. adding max volume id command for raft
2014-03-15 23:03:49 -07:00

74 lines
2.1 KiB
Go

package topology
import (
"code.google.com/p/weed-fs/go/glog"
"code.google.com/p/weed-fs/go/storage"
"math/rand"
"time"
)
func (t *Topology) StartRefreshWritableVolumes(garbageThreshold string) {
go func() {
for {
if t.IsLeader() {
freshThreshHold := time.Now().Unix() - 3*t.pulse //3 times of sleep interval
t.CollectDeadNodeAndFullVolumes(freshThreshHold, t.volumeSizeLimit)
}
time.Sleep(time.Duration(float32(t.pulse*1e3)*(1+rand.Float32())) * time.Millisecond)
}
}()
go func(garbageThreshold string) {
c := time.Tick(15 * time.Minute)
if t.IsLeader() {
for _ = range c {
t.Vacuum(garbageThreshold)
}
}
}(garbageThreshold)
go func() {
for {
select {
case v := <-t.chanFullVolumes:
t.SetVolumeCapacityFull(v)
case dn := <-t.chanRecoveredDataNodes:
t.RegisterRecoveredDataNode(dn)
glog.V(0).Infoln("DataNode", dn, "is back alive!")
case dn := <-t.chanDeadDataNodes:
t.UnRegisterDataNode(dn)
glog.V(0).Infoln("DataNode", dn, "is dead!")
}
}
}()
}
func (t *Topology) SetVolumeCapacityFull(volumeInfo storage.VolumeInfo) bool {
vl := t.GetVolumeLayout(volumeInfo.Collection, volumeInfo.ReplicaPlacement)
if !vl.SetVolumeCapacityFull(volumeInfo.Id) {
return false
}
for _, dn := range vl.vid2location[volumeInfo.Id].list {
if !volumeInfo.ReadOnly {
dn.UpAdjustActiveVolumeCountDelta(-1)
}
}
return true
}
func (t *Topology) UnRegisterDataNode(dn *DataNode) {
for _, v := range dn.volumes {
glog.V(0).Infoln("Removing Volume", v.Id, "from the dead volume server", dn)
vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement)
vl.SetVolumeUnavailable(dn, v.Id)
}
dn.UpAdjustVolumeCountDelta(-dn.GetVolumeCount())
dn.UpAdjustActiveVolumeCountDelta(-dn.GetActiveVolumeCount())
dn.UpAdjustMaxVolumeCountDelta(-dn.GetMaxVolumeCount())
dn.Parent().UnlinkChildNode(dn.Id())
}
func (t *Topology) RegisterRecoveredDataNode(dn *DataNode) {
for _, v := range dn.volumes {
vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement)
if vl.isWritable(&v) {
vl.SetVolumeAvailable(dn, v.Id)
}
}
}