dc and rack aware volume allocation

weed-fs/src/pkg/replication/volume_growth.go

@@ -1,6 +1,9 @@
 package replication
 
 import (
+	"fmt"
+	"math/rand"
+	"pkg/storage"
 	"pkg/topology"
 )
 
@@ -19,11 +22,74 @@ type VolumeGrowth struct {
 	copyAll     int
 }
 
-func (vg *VolumeGrowth) GrowVolumeCopy(copyLevel int, topo topology.Topology) {
-	if copyLevel == 1 {
-		for i := 0; i <vg.copy1factor; i++ {
-			topo.RandomlyReserveOneVolume()
+func (vg *VolumeGrowth) GrowVolumeCopy(copyLevel int, topo *topology.Topology) {
+	switch copyLevel {
+	case 1:
+		for i := 0; i < vg.copy1factor; i++ {
+			ret, server, vid := topo.RandomlyReserveOneVolume()
+			if ret {
+				vg.Grow(vid, server)
+			}
+		}
+	case 20:
+		for i := 0; i < vg.copy2factor; i++ {
+			nl := topology.NewNodeList(topo.Children(), nil)
+			picked, ret := nl.RandomlyPickN(2)
+			vid := topo.NextVolumeId()
+			if ret {
+				var servers []*topology.Server
+				for _, n := range picked {
+					if ok, server := n.ReserveOneVolume(rand.Intn(n.FreeSpace()), vid); ok {
+						servers = append(servers, server)
+					}
+				}
+				if len(servers) == 2 {
+					vg.Grow(vid, servers[0], servers[1])
+				}
+			}
+		}
+	case 30:
+		for i := 0; i < vg.copy3factor; i++ {
+			nl := topology.NewNodeList(topo.Children(), nil)
+			picked, ret := nl.RandomlyPickN(3)
+			vid := topo.NextVolumeId()
+			if ret {
+				var servers []*topology.Server
+				for _, n := range picked {
+					if ok, server := n.ReserveOneVolume(rand.Intn(n.FreeSpace()), vid); ok {
+						servers = append(servers, server)
+					}
+				}
+				if len(servers) == 3 {
+					vg.Grow(vid, servers[0], servers[1], servers[2])
+				}
+			}
+		}
+	case 02:
+		for i := 0; i < vg.copy2factor; i++ {
+			//randomly pick one server, and then choose from the same rack
+			ret, server1, vid := topo.RandomlyReserveOneVolume()
+			if ret {
+				rack := server1.Parent()
+				exclusion := make(map[string]topology.Node)
+				exclusion[server1.String()] = server1
+				newNodeList := topology.NewNodeList(rack.Children(), exclusion)
+				ret2, server2 := newNodeList.ReserveOneVolume(rand.Intn(newNodeList.FreeSpace()), vid)
+				if ret2 {
+					vg.Grow(vid, server1, server2)
+				}
+			}
+		}
+	case 12:
+		for i := 0; i < vg.copy3factor; i++ {
 		}
 	}
 
 }
+func (vg *VolumeGrowth) Grow(vid storage.VolumeId, servers ...*topology.Server) {
+	for _, server := range servers {
+		vi := &storage.VolumeInfo{Id: vid, Size: 0}
+		server.AddVolume(vi)
+	}
+	fmt.Println("Assigning", vid, "to", servers)
+}

weed-fs/src/pkg/replication/volume_growth_test.go

@@ -0,0 +1,129 @@
+package replication
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"pkg/storage"
+  "pkg/topology"
+	"testing"
+	"time"
+)
+
+var topologyLayout = `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":12312},
+          {"id":2, "size":12312},
+          {"id":3, "size":12312}
+        ],
+        "limit":3
+      },
+      "server2":{
+        "volumes":[
+          {"id":4, "size":12312},
+          {"id":5, "size":12312},
+          {"id":6, "size":12312}
+        ],
+        "limit":10
+      }
+    },
+    "rack2":{
+      "server1":{
+        "volumes":[
+          {"id":4, "size":12312},
+          {"id":5, "size":12312},
+          {"id":6, "size":12312}
+        ],
+        "limit":4
+      },
+      "server2":{
+        "volumes":[],
+        "limit":4
+      },
+      "server3":{
+        "volumes":[
+          {"id":2, "size":12312},
+          {"id":3, "size":12312},
+          {"id":4, "size":12312}
+        ],
+        "limit":2
+      }
+    }
+  },
+  "dc2":{
+  },
+  "dc3":{
+    "rack2":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":12312},
+          {"id":3, "size":12312},
+          {"id":5, "size":12312}
+        ],
+        "limit":4
+      }
+    }
+  }
+}
+`
+
+func setup(topologyLayout string) *topology.Topology {
+	var data interface{}
+	err := json.Unmarshal([]byte(topologyLayout), &data)
+	if err != nil {
+		fmt.Println("error:", err)
+	}
+	fmt.Println("data:", data)
+
+	//need to connect all nodes first before server adding volumes
+	topo := topology.NewTopology("mynetwork")
+	mTopology := data.(map[string]interface{})
+	for dcKey, dcValue := range mTopology {
+		dc := topology.NewDataCenter(dcKey)
+		dcMap := dcValue.(map[string]interface{})
+		topo.LinkChildNode(dc)
+		for rackKey, rackValue := range dcMap {
+			rack := topology.NewRack(rackKey)
+			rackMap := rackValue.(map[string]interface{})
+			dc.LinkChildNode(rack)
+			for serverKey, serverValue := range rackMap {
+				server := topology.NewServer(serverKey)
+				serverMap := serverValue.(map[string]interface{})
+				rack.LinkChildNode(server)
+				for _, v := range serverMap["volumes"].([]interface{}) {
+					m := v.(map[string]interface{})
+					vi := &storage.VolumeInfo{Id: storage.VolumeId(int64(m["id"].(float64))), Size: int64(m["size"].(float64))}
+					server.AddVolume(vi)
+				}
+				server.UpAdjustMaxVolumeCountDelta(int(serverMap["limit"].(float64)))
+			}
+		}
+	}
+
+	fmt.Println("topology:", *topo)
+
+	return topo
+}
+
+func TestRemoveDataCenter(t *testing.T) {
+	topo := setup(topologyLayout)
+	topo.UnlinkChildNode(topology.NodeId("dc2"))
+	if topo.GetActiveVolumeCount() != 15 {
+		t.Fail()
+	}
+	topo.UnlinkChildNode(topology.NodeId("dc3"))
+	if topo.GetActiveVolumeCount() != 12 {
+		t.Fail()
+	}
+}
+
+func TestReserveOneVolume(t *testing.T) {
+	topo := setup(topologyLayout)
+  rand.Seed(time.Now().UnixNano())
+  vg:=&VolumeGrowth{copy1factor:3,copy2factor:2,copy3factor:1,copyAll:4}
+  vg.GrowVolumeCopy(20,topo)
+}

weed-fs/src/pkg/topology/node.go

@@ -20,8 +20,10 @@ type Node interface {
 	setParent(Node)
 	LinkChildNode(node Node)
 	UnlinkChildNode(nodeId NodeId)
-	
+
 	IsServer() bool
+	Children() map[NodeId]Node
+  Parent() Node
 }
 type NodeImpl struct {
 	id                NodeId
@@ -59,22 +61,28 @@ func (n *NodeImpl) FreeSpace() int {
 func (n *NodeImpl) setParent(node Node) {
 	n.parent = node
 }
+func (n *NodeImpl) Children() map[NodeId]Node {
+	return n.children
+}
+func (n *NodeImpl) Parent() Node {
+  return n.parent
+}
 func (n *NodeImpl) ReserveOneVolume(r int, vid storage.VolumeId) (bool, *Server) {
 	ret := false
 	var assignedNode *Server
 	for _, node := range n.children {
 		freeSpace := node.FreeSpace()
-		fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace)
+		//fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace)
 		if freeSpace <= 0 {
 			continue
 		}
 		if r >= freeSpace {
 			r -= freeSpace
 		} else {
-		  if node.IsServer() && node.FreeSpace()>0 {
-		    fmt.Println("vid =", vid, " assigned to node =", node, ", freeSpace =", node.FreeSpace())
-		    return true, node.(*Server)
-		  }
+			if node.IsServer() && node.FreeSpace() > 0 {
+				//fmt.Println("vid =", vid, " assigned to node =", node, ", freeSpace =", node.FreeSpace())
+				return true, node.(*Server)
+			}
 			ret, assignedNode = node.ReserveOneVolume(r, vid)
 			if ret {
 				break

weed-fs/src/pkg/topology/node_list.go

@@ -0,0 +1,69 @@
+package topology
+
+import (
+	"fmt"
+	"math/rand"
+	"pkg/storage"
+)
+
+type NodeList struct {
+	nodes  map[NodeId]Node
+	except map[string]Node
+}
+
+func NewNodeList(nodes map[NodeId]Node, except map[string]Node) *NodeList {
+	m := make(map[NodeId]Node, len(nodes)-len(except))
+	for _, n := range nodes {
+		if except[n.String()] == nil {
+			m[n.Id()] = n
+		}
+	}
+	nl := &NodeList{nodes: m}
+	return nl
+}
+
+func (nl *NodeList) FreeSpace() int {
+	freeSpace := 0
+	for _, n := range nl.nodes {
+		freeSpace += n.FreeSpace()
+	}
+	return freeSpace
+}
+
+func (nl *NodeList) RandomlyPickN(n int) ([]Node, bool) {
+	var list []Node
+	for _, n := range nl.nodes {
+		if n.FreeSpace() > 0 {
+			list = append(list, n)
+		}
+	}
+	if n > len(list){
+	  return nil,false
+	}
+	for i := n; i > 0; i-- {
+	  r := rand.Intn(i)
+	  t := list[r]
+	  list[r] = list[i-1]
+	  list[i-1] = t
+	}
+	return list[len(list)-n:], true
+}
+
+func (nl *NodeList) ReserveOneVolume(randomVolumeIndex int, vid storage.VolumeId) (bool, *Server) {
+	for _, node := range nl.nodes {
+		freeSpace := node.FreeSpace()
+		if randomVolumeIndex >= freeSpace {
+			randomVolumeIndex -= freeSpace
+		} else {
+			if node.IsServer() && node.FreeSpace() > 0 {
+				fmt.Println("vid =", vid, " assigned to node =", node, ", freeSpace =", node.FreeSpace())
+				return true, node.(*Server)
+			}
+			children := node.Children()
+			newNodeList := NewNodeList(children, nl.except)
+			return newNodeList.ReserveOneVolume(randomVolumeIndex, vid)
+		}
+	}
+	return false, nil
+
+}

weed-fs/src/pkg/topology/node_list_test.go

@@ -0,0 +1,39 @@
+package topology
+
+import (
+	"strconv"
+	"testing"
+  _ "fmt"
+)
+
+func TestXYZ(t *testing.T) {
+	topo := NewTopology("topo")
+	for i := 0; i < 5; i++ {
+		dc := NewDataCenter("dc" + strconv.Itoa(i))
+		dc.activeVolumeCount = i
+		dc.maxVolumeCount = 5
+		topo.LinkChildNode(dc)
+	}
+	nl := NewNodeList(topo.Children(),nil)
+
+  picked, ret := nl.RandomlyPickN(1)
+  if !ret || len(picked)!=1 {
+    t.Errorf("need to randomly pick 1 node")
+  }
+
+	picked, ret = nl.RandomlyPickN(4)
+	if !ret || len(picked)!=4 {
+	  t.Errorf("need to randomly pick 4 nodes")
+	}
+
+  picked, ret = nl.RandomlyPickN(5)
+  if !ret || len(picked)!=5 {
+    t.Errorf("need to randomly pick 5 nodes")
+  }
+
+  picked, ret = nl.RandomlyPickN(6)
+  if ret || len(picked)!=0 {
+    t.Errorf("can not randomly pick 6 nodes:", ret, picked)
+  }
+
+}

weed-fs/src/pkg/topology/topo_test.go

@@ -76,8 +76,8 @@ func setup(topologyLayout string) *Topology {
 	if err != nil {
 		fmt.Println("error:", err)
 	}
-	fmt.Println("data:", data)
-	printMap(data)
+	//fmt.Println("data:", data)
+	//printMap(data)
 
 	//need to connect all nodes first before server adding volumes
 	topo := NewTopology("mynetwork")
@@ -156,4 +156,5 @@ func TestReserveOneVolume(t *testing.T) {
 	ret, node, vid := topo.RandomlyReserveOneVolume()
 	fmt.Println("topology:", topo)
   fmt.Println("assigned :", ret, ", node :", node,", volume id:", vid)
+
 }

weed-fs/src/pkg/topology/topology.go

@@ -1,7 +1,7 @@
 package topology
 
 import (
-	"fmt"
+	_ "fmt"
 	"math/rand"
 	"pkg/storage"
 )
@@ -18,13 +18,22 @@ func NewTopology(id string) *Topology {
 	return t
 }
 func (t *Topology) RandomlyReserveOneVolume() (bool, *Server, storage.VolumeId) {
-	vid := t.nextVolumeId()
+	vid := t.NextVolumeId()
 	ret, node := t.ReserveOneVolume(rand.Intn(t.FreeSpace()), vid)
-  fmt.Println("node.IsServer", node.IsServer())
   return ret, node, vid
 }
 
-func (t *Topology) nextVolumeId() storage.VolumeId {
+func (t *Topology) RandomlyReserveOneVolumeExcept(except []Node) (bool, *Server, storage.VolumeId) {
+  freeSpace := t.FreeSpace()
+  for _, node := range except {
+    freeSpace -= node.FreeSpace()
+  }
+  vid := t.NextVolumeId()
+  ret, node := t.ReserveOneVolume(rand.Intn(freeSpace), vid)
+  return ret, node, vid
+}
+
+func (t *Topology) NextVolumeId() storage.VolumeId {
 	vid := t.GetMaxVolumeId()
 	return vid.Next()
 }