gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

interface and default implementation for node

Browse source
This commit is contained in:
Chris Lu 2012-09-02 11:36:23 -07:00
parent a467d5081c
commit 6a84f50d24
6 changed files with 148 additions and 111 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
weed-fs/src/pkg/topology/data_center.go
View file

@ -4,12 +4,13 @@ import (
) )
type DataCenter struct { type DataCenter struct {
Node NodeImpl
ipRange IpRange ipRange IpRange
} }
func NewDataCenter(id NodeId) *DataCenter{ func NewDataCenter(id string) *DataCenter{
dc := &DataCenter{} dc := &DataCenter{}
dc.Node = *NewNode() dc.id = NodeId(id)
dc.Node.Id = id dc.nodeType = "DataCenter"
dc.children = make(map[NodeId]Node)
return dc return dc
} }

128
weed-fs/src/pkg/topology/node.go
View file

@ -6,35 +6,66 @@ import (
) )
type NodeId string type NodeId string
type Node struct { type Node interface {
Id NodeId Id() NodeId
String() string
FreeSpace() int
ReserveOneVolume(r int, vid storage.VolumeId) (bool, Node)
UpAdjustMaxVolumeCountDelta(maxVolumeCountDelta int)
UpAdjustActiveVolumeCountDelta(activeVolumeCountDelta int)
UpAdjustMaxVolumeId(vid storage.VolumeId)
GetActiveVolumeCount() int
GetMaxVolumeCount() int
GetMaxVolumeId() storage.VolumeId
setParent(Node)
LinkChildNode(node Node)
UnlinkChildNode(nodeId NodeId)
}
type NodeImpl struct {
id NodeId
activeVolumeCount int activeVolumeCount int
maxVolumeCount int maxVolumeCount int
parent *Node parent Node
children map[NodeId]*Node children map[NodeId]Node
maxVolumeId storage.VolumeId maxVolumeId storage.VolumeId
//for rack, data center, topology
nodeType string
} }
func NewNode() *Node { func (n *NodeImpl) IsServer() bool {
n := &Node{} return n.nodeType == "Server"
n.children = make(map[NodeId]*Node)
return n
} }
func (n *Node) String() string { func (n *NodeImpl) IsRack() bool {
return n.nodeType == "Rack"
}
func (n *NodeImpl) IsDataCenter() bool {
return n.nodeType == "DataCenter"
}
func (n *NodeImpl) String() string {
if n.parent != nil { if n.parent != nil {
return n.parent.String()+":"+string(n.Id) return n.parent.String() + ":" + string(n.id)
} }
return string(n.Id) return string(n.id)
} }
func (n *NodeImpl) Id() NodeId {
func (n *Node) ReserveOneVolume(r int, vid storage.VolumeId) (bool, *Node) { return n.id
if n.children == nil { }
func (n *NodeImpl) FreeSpace() int {
return n.maxVolumeCount - n.activeVolumeCount
}
func (n *NodeImpl) setParent(node Node) {
n.parent = node
}
func (n *NodeImpl) ReserveOneVolume(r int, vid storage.VolumeId) (bool, Node) {
if n.IsServer() && n.maxVolumeCount > n.activeVolumeCount {
fmt.Println("vid =", vid, " assigned to node =", n, ", freeSpace =", n.maxVolumeCount-n.activeVolumeCount)
return true, n return true, n
} }
ret := false ret := false
var assignedNode *Node var assignedNode Node
for _, node := range n.children { for _, node := range n.children {
freeSpace := node.maxVolumeCount - node.activeVolumeCount freeSpace := node.FreeSpace()
fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace) fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace)
if freeSpace <= 0 { if freeSpace <= 0 {
continue continue
@ -51,48 +82,57 @@ func (n *Node) ReserveOneVolume(r int, vid storage.VolumeId) (bool, *Node) {
return ret, assignedNode return ret, assignedNode
} }
func (n *Node) AddVolume(v *storage.VolumeInfo) { func (n *NodeImpl) UpAdjustMaxVolumeCountDelta(maxVolumeCountDelta int) { //can be negative
if n.maxVolumeId < v.Id { n.maxVolumeCount += maxVolumeCountDelta
n.maxVolumeId = v.Id
}
n.activeVolumeCount++
fmt.Println(n.Id, "adds 1, volumeCount =", n.activeVolumeCount)
if n.parent != nil { if n.parent != nil {
n.parent.AddVolume(v) n.parent.UpAdjustMaxVolumeCountDelta(maxVolumeCountDelta)
} }
} }
func (n *Node) AddMaxVolumeCount(maxVolumeCount int) {//can be negative func (n *NodeImpl) UpAdjustActiveVolumeCountDelta(activeVolumeCountDelta int) { //can be negative
n.maxVolumeCount += maxVolumeCount n.activeVolumeCount += activeVolumeCountDelta
if n.parent != nil { if n.parent != nil {
n.parent.AddMaxVolumeCount(maxVolumeCount) n.parent.UpAdjustActiveVolumeCountDelta(activeVolumeCountDelta)
}
}
func (n *NodeImpl) UpAdjustMaxVolumeId(vid storage.VolumeId) { //can be negative
if n.maxVolumeId < vid {
n.maxVolumeId = vid
if n.parent != nil {
n.parent.UpAdjustMaxVolumeId(vid)
}
} }
} }
func (n *Node) GetMaxVolumeId() storage.VolumeId { func (n *NodeImpl) GetMaxVolumeId() storage.VolumeId {
return n.maxVolumeId return n.maxVolumeId
} }
func (n *NodeImpl) GetActiveVolumeCount() int {
return n.activeVolumeCount
}
func (n *NodeImpl) GetMaxVolumeCount() int {
return n.maxVolumeCount
}
func (n *Node) AddNode(node *Node) { func (n *NodeImpl) LinkChildNode(node Node) {
if n.children[node.Id] == nil { if n.children[node.Id()] == nil {
n.children[node.Id] = node n.children[node.Id()] = node
n.activeVolumeCount += node.activeVolumeCount n.activeVolumeCount += node.GetActiveVolumeCount()
n.maxVolumeCount += node.maxVolumeCount n.maxVolumeCount += node.GetMaxVolumeCount()
fmt.Println(n.Id, "adds", node.Id, "volumeCount =", n.activeVolumeCount) node.setParent(n)
if n.maxVolumeId < node.GetMaxVolumeId() {
n.maxVolumeId = node.GetMaxVolumeId()
}
fmt.Println(n, "adds", node, "volumeCount =", n.activeVolumeCount)
} }
} }
func (n *Node) RemoveNode(nodeId NodeId) { func (n *NodeImpl) UnlinkChildNode(nodeId NodeId) {
node := n.children[nodeId] node := n.children[nodeId]
node.setParent(nil)
if node != nil { if node != nil {
delete(n.children, node.Id) delete(n.children, node.Id())
n.activeVolumeCount -= node.activeVolumeCount n.UpAdjustActiveVolumeCountDelta(-node.GetActiveVolumeCount())
n.maxVolumeCount -= node.maxVolumeCount n.UpAdjustMaxVolumeCountDelta(-node.GetMaxVolumeCount())
p := n.parent fmt.Println(n, "removes", node, "volumeCount =", n.activeVolumeCount)
for p != nil {
p.activeVolumeCount -= node.activeVolumeCount
p.maxVolumeCount -= node.maxVolumeCount
p = p.parent
}
fmt.Println(n.Id, "removes", node.Id, "volumeCount =", n.activeVolumeCount)
} }
} }

9
weed-fs/src/pkg/topology/rack.go
View file

@ -4,13 +4,14 @@ import (
) )
type Rack struct { type Rack struct {
Node NodeImpl
ipRange IpRange ipRange IpRange
} }
func NewRack(id NodeId) *Rack { func NewRack(id string) *Rack {
r := &Rack{} r := &Rack{}
r.Node = *NewNode() r.id = NodeId(id)
r.Node.Id = id r.nodeType = "Rack"
r.children = make(map[NodeId]Node)
return r return r
} }

10
weed-fs/src/pkg/topology/server.go
View file

@ -6,15 +6,16 @@ import (
) )
type Server struct { type Server struct {
Node NodeImpl
volumes map[storage.VolumeId]*storage.VolumeInfo volumes map[storage.VolumeId]*storage.VolumeInfo
Ip NodeId Ip NodeId
Port int Port int
PublicUrl string PublicUrl string
} }
func NewServer(id NodeId) *Server{ func NewServer(id string) *Server{
s := &Server{} s := &Server{}
s.Node.Id = id s.id = NodeId(id)
s.nodeType = "Server"
s.volumes = make(map[storage.VolumeId]*storage.VolumeInfo) s.volumes = make(map[storage.VolumeId]*storage.VolumeInfo)
return s return s
} }
@ -24,5 +25,6 @@ func (s *Server) CreateOneVolume(r int, vid storage.VolumeId) storage.VolumeId {
} }
func (s *Server) AddVolume(v *storage.VolumeInfo){ func (s *Server) AddVolume(v *storage.VolumeInfo){
s.volumes[v.Id] = v s.volumes[v.Id] = v
s.Node.AddVolume(v) s.UpAdjustActiveVolumeCountDelta(1)
s.UpAdjustMaxVolumeId(v.Id)
} }

38
weed-fs/src/pkg/topology/topo_test.go
View file

@ -80,36 +80,33 @@ func setup() *Topology {
printMap(data) printMap(data)
//need to connect all nodes first before server adding volumes //need to connect all nodes first before server adding volumes
topo := NewTopology(NodeId("mynetwork")) topo := NewTopology("mynetwork")
mTopology := data.(map[string]interface{}) mTopology := data.(map[string]interface{})
for dcKey, dcValue := range mTopology { for dcKey, dcValue := range mTopology {
dc := NewDataCenter(NodeId(dcKey)) dc := NewDataCenter(dcKey)
dc.Node.parent = &topo.Node
dcMap := dcValue.(map[string]interface{}) dcMap := dcValue.(map[string]interface{})
topo.Node.AddNode(&dc.Node) topo.LinkChildNode(dc)
for rackKey, rackValue := range dcMap { for rackKey, rackValue := range dcMap {
rack := NewRack(NodeId(rackKey)) rack := NewRack(rackKey)
rack.Node.parent = &dc.Node
rackMap := rackValue.(map[string]interface{}) rackMap := rackValue.(map[string]interface{})
dc.Node.AddNode(&rack.Node) dc.LinkChildNode(rack)
for serverKey, serverValue := range rackMap { for serverKey, serverValue := range rackMap {
server := NewServer(NodeId(serverKey)) server := NewServer(serverKey)
server.Node.parent = &rack.Node
serverMap := serverValue.(map[string]interface{}) serverMap := serverValue.(map[string]interface{})
rack.Node.AddNode(&server.Node) rack.LinkChildNode(server)
for _, v := range serverMap["volumes"].([]interface{}) { for _, v := range serverMap["volumes"].([]interface{}) {
m := v.(map[string]interface{}) m := v.(map[string]interface{})
vi := &storage.VolumeInfo{Id: storage.VolumeId(int64(m["id"].(float64))), Size: int64(m["size"].(float64))} vi := &storage.VolumeInfo{Id: storage.VolumeId(int64(m["id"].(float64))), Size: int64(m["size"].(float64))}
server.AddVolume(vi) server.AddVolume(vi)
} }
server.Node.AddMaxVolumeCount(int(serverMap["limit"].(float64))) server.UpAdjustMaxVolumeCountDelta(int(serverMap["limit"].(float64)))
} }
} }
} }
fmt.Println("topology:", *topo) fmt.Println("topology:", *topo)
bytes, err := json.Marshal(topo.Node.children) bytes, err := json.Marshal(topo.children)
if err != nil { if err != nil {
fmt.Println("json error:", err) fmt.Println("json error:", err)
} }
@ -140,20 +137,14 @@ func printMap(mm interface{}) {
} }
} }
func TestAddVolume(t *testing.T) {
topo := setup()
v := &storage.VolumeInfo{}
topo.AddVolume(v)
}
func TestRemoveDataCenter(t *testing.T) { func TestRemoveDataCenter(t *testing.T) {
topo := setup() topo := setup()
topo.RemoveNode(NodeId("dc2")) topo.UnlinkChildNode(NodeId("dc2"))
if topo.activeVolumeCount != 15 { if topo.GetActiveVolumeCount() != 15 {
t.Fail() t.Fail()
} }
topo.RemoveNode(NodeId("dc3")) topo.UnlinkChildNode(NodeId("dc3"))
if topo.activeVolumeCount != 12 { if topo.GetActiveVolumeCount() != 12 {
t.Fail() t.Fail()
} }
} }
@ -161,8 +152,9 @@ func TestRemoveDataCenter(t *testing.T) {
func TestReserveOneVolume(t *testing.T) { func TestReserveOneVolume(t *testing.T) {
topo := setup() topo := setup()
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
rand.Seed(1)
ret, node, vid := topo.RandomlyReserveOneVolume() ret, node, vid := topo.RandomlyReserveOneVolume()
fmt.Println("topology:", topo.Node) fmt.Println("topology:", topo)
fmt.Println("assigned :", ret) fmt.Println("assigned :", ret)
fmt.Println("assigned node :", node) fmt.Println("assigned node :", node)
fmt.Println("assigned volume id:", vid) fmt.Println("assigned volume id:", vid)

23
weed-fs/src/pkg/topology/topology.go
View file

@ -7,24 +7,25 @@ import (
) )
type Topology struct { type Topology struct {
Node NodeImpl
}
func NewTopology(id NodeId) *Topology{
t := &Topology{}
t.Node = *NewNode()
t.Node.Id = id
return t
} }
func (t *Topology) RandomlyReserveOneVolume() (bool, *Node, storage.VolumeId) { func NewTopology(id string) *Topology {
slots := t.Node.maxVolumeCount-t.Node.activeVolumeCount t := &Topology{}
t.id = NodeId(id)
t.nodeType = "Topology"
t.children = make(map[NodeId]Node)
return t
}
func (t *Topology) RandomlyReserveOneVolume() (bool, Node, storage.VolumeId) {
slots := t.maxVolumeCount-t.activeVolumeCount
r := rand.Intn(slots) r := rand.Intn(slots)
vid := t.nextVolumeId() vid := t.nextVolumeId()
ret, node := t.Node.ReserveOneVolume(r,vid) ret, node := t.ReserveOneVolume(r,vid)
return ret, node, vid return ret, node, vid
} }
func (t *Topology) nextVolumeId() storage.VolumeId { func (t *Topology) nextVolumeId() storage.VolumeId {
vid := t.Node.GetMaxVolumeId() vid := t.GetMaxVolumeId()
return vid.Next() return vid.Next()
} }