2012-08-24 05:33:37 +00:00
|
|
|
package topology
|
|
|
|
|
|
|
|
import (
|
2012-09-17 00:31:15 +00:00
|
|
|
"errors"
|
2012-09-24 09:01:53 +00:00
|
|
|
"io/ioutil"
|
2012-09-02 21:33:48 +00:00
|
|
|
"math/rand"
|
2012-09-17 00:31:15 +00:00
|
|
|
"pkg/directory"
|
2012-09-10 07:18:07 +00:00
|
|
|
"pkg/sequence"
|
2012-08-28 08:04:39 +00:00
|
|
|
"pkg/storage"
|
2012-08-24 05:33:37 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
type Topology struct {
|
2012-09-02 21:33:48 +00:00
|
|
|
NodeImpl
|
2012-09-10 07:18:07 +00:00
|
|
|
|
|
|
|
//transient vid~servers mapping for each replication type
|
|
|
|
replicaType2VolumeLayout []*VolumeLayout
|
|
|
|
|
|
|
|
pulse int64
|
|
|
|
|
|
|
|
volumeSizeLimit uint64
|
|
|
|
|
|
|
|
sequence sequence.Sequencer
|
2012-09-18 21:05:12 +00:00
|
|
|
|
|
|
|
chanDeadDataNodes chan *DataNode
|
|
|
|
chanRecoveredDataNodes chan *DataNode
|
2012-12-04 04:28:12 +00:00
|
|
|
chanFullVolumes chan storage.VolumeInfo
|
2012-09-24 09:01:53 +00:00
|
|
|
|
2012-09-23 03:46:31 +00:00
|
|
|
configuration *Configuration
|
2012-08-31 08:35:11 +00:00
|
|
|
}
|
2012-08-28 08:04:39 +00:00
|
|
|
|
2012-09-24 09:01:53 +00:00
|
|
|
func NewTopology(id string, confFile string, dirname string, sequenceFilename string, volumeSizeLimit uint64, pulse int) *Topology {
|
2012-09-02 21:33:48 +00:00
|
|
|
t := &Topology{}
|
|
|
|
t.id = NodeId(id)
|
|
|
|
t.nodeType = "Topology"
|
2012-09-19 08:45:30 +00:00
|
|
|
t.NodeImpl.value = t
|
2012-09-02 21:33:48 +00:00
|
|
|
t.children = make(map[NodeId]Node)
|
2012-09-10 07:18:07 +00:00
|
|
|
t.replicaType2VolumeLayout = make([]*VolumeLayout, storage.LengthRelicationType)
|
|
|
|
t.pulse = int64(pulse)
|
|
|
|
t.volumeSizeLimit = volumeSizeLimit
|
2012-09-19 08:45:30 +00:00
|
|
|
|
2012-09-24 09:01:53 +00:00
|
|
|
t.sequence = sequence.NewSequencer(dirname, sequenceFilename)
|
2012-09-19 08:45:30 +00:00
|
|
|
|
2012-09-18 21:05:12 +00:00
|
|
|
t.chanDeadDataNodes = make(chan *DataNode)
|
|
|
|
t.chanRecoveredDataNodes = make(chan *DataNode)
|
2012-12-04 04:28:12 +00:00
|
|
|
t.chanFullVolumes = make(chan storage.VolumeInfo)
|
2012-09-19 08:45:30 +00:00
|
|
|
|
2012-09-24 09:01:53 +00:00
|
|
|
t.loadConfiguration(confFile)
|
|
|
|
|
2012-09-02 21:33:48 +00:00
|
|
|
return t
|
|
|
|
}
|
2012-09-17 00:31:15 +00:00
|
|
|
|
2012-09-24 09:01:53 +00:00
|
|
|
func (t *Topology) loadConfiguration(configurationFile string) error {
|
|
|
|
b, e := ioutil.ReadFile(configurationFile)
|
|
|
|
if e == nil {
|
|
|
|
t.configuration, e = NewConfiguration(b)
|
|
|
|
}
|
|
|
|
return e
|
2012-09-23 03:46:31 +00:00
|
|
|
}
|
|
|
|
|
2012-09-24 09:01:53 +00:00
|
|
|
func (t *Topology) Lookup(vid storage.VolumeId) *[]*DataNode {
|
|
|
|
for _, vl := range t.replicaType2VolumeLayout {
|
|
|
|
if vl != nil {
|
|
|
|
if list := vl.Lookup(vid); list != nil {
|
|
|
|
return list
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
2012-09-23 22:45:26 +00:00
|
|
|
}
|
|
|
|
|
2012-09-17 00:31:15 +00:00
|
|
|
func (t *Topology) RandomlyReserveOneVolume() (bool, *DataNode, *storage.VolumeId) {
|
2012-09-17 02:18:37 +00:00
|
|
|
if t.FreeSpace() <= 0 {
|
|
|
|
return false, nil, nil
|
|
|
|
}
|
2012-09-03 08:50:04 +00:00
|
|
|
vid := t.NextVolumeId()
|
2012-09-02 21:33:48 +00:00
|
|
|
ret, node := t.ReserveOneVolume(rand.Intn(t.FreeSpace()), vid)
|
2012-09-17 00:31:15 +00:00
|
|
|
return ret, node, &vid
|
2012-08-28 08:04:39 +00:00
|
|
|
}
|
2012-08-29 07:58:03 +00:00
|
|
|
|
2012-09-17 02:18:37 +00:00
|
|
|
func (t *Topology) RandomlyReserveOneVolumeExcept(except []Node) (bool, *DataNode, *storage.VolumeId) {
|
2012-09-10 07:18:07 +00:00
|
|
|
freeSpace := t.FreeSpace()
|
|
|
|
for _, node := range except {
|
|
|
|
freeSpace -= node.FreeSpace()
|
|
|
|
}
|
2012-09-17 02:18:37 +00:00
|
|
|
if freeSpace <= 0 {
|
|
|
|
return false, nil, nil
|
|
|
|
}
|
2012-09-10 07:18:07 +00:00
|
|
|
vid := t.NextVolumeId()
|
|
|
|
ret, node := t.ReserveOneVolume(rand.Intn(freeSpace), vid)
|
2012-09-17 02:18:37 +00:00
|
|
|
return ret, node, &vid
|
2012-09-03 08:50:04 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (t *Topology) NextVolumeId() storage.VolumeId {
|
2012-09-02 21:33:48 +00:00
|
|
|
vid := t.GetMaxVolumeId()
|
2012-08-29 08:37:40 +00:00
|
|
|
return vid.Next()
|
2012-08-28 08:04:39 +00:00
|
|
|
}
|
2012-09-10 07:18:07 +00:00
|
|
|
|
2012-09-17 00:31:15 +00:00
|
|
|
func (t *Topology) PickForWrite(repType storage.ReplicationType, count int) (string, int, *DataNode, error) {
|
2012-09-17 06:18:47 +00:00
|
|
|
replicationTypeIndex := repType.GetReplicationLevelIndex()
|
2012-09-17 00:31:15 +00:00
|
|
|
if t.replicaType2VolumeLayout[replicationTypeIndex] == nil {
|
|
|
|
t.replicaType2VolumeLayout[replicationTypeIndex] = NewVolumeLayout(repType, t.volumeSizeLimit, t.pulse)
|
|
|
|
}
|
|
|
|
vid, count, datanodes, err := t.replicaType2VolumeLayout[replicationTypeIndex].PickForWrite(count)
|
|
|
|
if err != nil {
|
|
|
|
return "", 0, nil, errors.New("No writable volumes avalable!")
|
|
|
|
}
|
|
|
|
fileId, count := t.sequence.NextFileId(count)
|
|
|
|
return directory.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes.Head(), nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (t *Topology) GetVolumeLayout(repType storage.ReplicationType) *VolumeLayout {
|
2012-09-17 06:18:47 +00:00
|
|
|
replicationTypeIndex := repType.GetReplicationLevelIndex()
|
2012-09-17 00:31:15 +00:00
|
|
|
if t.replicaType2VolumeLayout[replicationTypeIndex] == nil {
|
|
|
|
t.replicaType2VolumeLayout[replicationTypeIndex] = NewVolumeLayout(repType, t.volumeSizeLimit, t.pulse)
|
|
|
|
}
|
|
|
|
return t.replicaType2VolumeLayout[replicationTypeIndex]
|
2012-09-14 08:17:13 +00:00
|
|
|
}
|
|
|
|
|
2012-09-17 00:31:15 +00:00
|
|
|
func (t *Topology) RegisterVolumeLayout(v *storage.VolumeInfo, dn *DataNode) {
|
|
|
|
t.GetVolumeLayout(v.RepType).RegisterVolume(v, dn)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (t *Topology) RegisterVolumes(volumeInfos []storage.VolumeInfo, ip string, port int, publicUrl string, maxVolumeCount int) {
|
2012-09-24 09:01:53 +00:00
|
|
|
dcName, rackName := t.configuration.Locate(ip)
|
2012-09-23 03:46:31 +00:00
|
|
|
dc := t.GetOrCreateDataCenter(dcName)
|
|
|
|
rack := dc.GetOrCreateRack(rackName)
|
2012-09-17 00:31:15 +00:00
|
|
|
dn := rack.GetOrCreateDataNode(ip, port, publicUrl, maxVolumeCount)
|
|
|
|
for _, v := range volumeInfos {
|
2012-09-19 08:45:30 +00:00
|
|
|
dn.AddOrUpdateVolume(v)
|
2012-09-17 00:31:15 +00:00
|
|
|
t.RegisterVolumeLayout(&v, dn)
|
|
|
|
}
|
2012-09-14 08:17:13 +00:00
|
|
|
}
|
|
|
|
|
2012-09-23 03:46:31 +00:00
|
|
|
func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter {
|
2012-09-17 00:31:15 +00:00
|
|
|
for _, c := range t.Children() {
|
|
|
|
dc := c.(*DataCenter)
|
2012-09-23 03:46:31 +00:00
|
|
|
if string(dc.Id()) == dcName {
|
2012-09-17 00:31:15 +00:00
|
|
|
return dc
|
|
|
|
}
|
|
|
|
}
|
2012-09-23 03:46:31 +00:00
|
|
|
dc := NewDataCenter(dcName)
|
2012-09-17 00:31:15 +00:00
|
|
|
t.LinkChildNode(dc)
|
|
|
|
return dc
|
|
|
|
}
|
|
|
|
|