adjusting refresh topology writable volumes(not finished yet)

Need to refreshWritableVolumes for each replication type

7 files changed, 27 insertions, 7 deletions

diff --git a/weed-fs/src/pkg/directory/volume_mapping.go b/weed-fs/src/pkg/directory/volume_mapping.go
index 495c57930..da41b3510 100644
--- a/weed-fs/src/pkg/directory/volume_mapping.go
+++ b/weed-fs/src/pkg/directory/volume_mapping.go
@@ -120,7 +120,7 @@ func (m *Mapper) StartRefreshWritableVolumes() {
 }
 
 func (m *Mapper) refreshWritableVolumes() {
-	freshThreshHold := time.Now().Unix() - 5*m.pulse //5 times of sleep interval
+	freshThreshHold := time.Now().Unix() - 3*m.pulse //3 times of sleep interval
 	//setting Writers, copy-on-write because of possible updating, this needs some future work!
 	var writers []storage.VolumeId
 	for _, machine_entry := range m.Machines {
diff --git a/weed-fs/src/pkg/replication/volume_growth.go b/weed-fs/src/pkg/replication/volume_growth.go
index 8519921f5..28dca68b4 100644
--- a/weed-fs/src/pkg/replication/volume_growth.go
+++ b/weed-fs/src/pkg/replication/volume_growth.go
@@ -44,7 +44,7 @@ func (vg *VolumeGrowth) GrowByType(repType storage.ReplicationType, topo *topolo
 	case storage.Copy11:
 		return vg.GrowByCountAndType(vg.copy3factor, repType, topo)
 	}
-	return 0, nil
+	return 0, errors.New("Unknown Replication Type!")
 }
 func (vg *VolumeGrowth) GrowByCountAndType(count int, repType storage.ReplicationType, topo *topology.Topology) (counter int, err error) {
 	counter = 0
diff --git a/weed-fs/src/pkg/topology/data_node.go b/weed-fs/src/pkg/topology/data_node.go
index 1516572fd..04c7cf111 100644
--- a/weed-fs/src/pkg/topology/data_node.go
+++ b/weed-fs/src/pkg/topology/data_node.go
@@ -11,7 +11,7 @@ type DataNode struct {
 	Ip        string
 	Port      int
 	PublicUrl string
-	lastSeen  int64 // unix time in seconds
+	LastSeen  int64 // unix time in seconds
 }
 
 func NewDataNode(id string) *DataNode {
diff --git a/weed-fs/src/pkg/topology/node.go b/weed-fs/src/pkg/topology/node.go
index e73719794..ddaf9f0b2 100644
--- a/weed-fs/src/pkg/topology/node.go
+++ b/weed-fs/src/pkg/topology/node.go
@@ -151,7 +151,7 @@ func (n *NodeImpl) CollectWritableVolumes(freshThreshHold int64, volumeSizeLimit
 	if n.IsRack() {
 		for _, c := range n.Children() {
 			dn := c.(*DataNode) //can not cast n to DataNode
-			if dn.lastSeen > freshThreshHold {
+			if dn.LastSeen > freshThreshHold {
 			  continue
 			}
 			for _, v := range dn.volumes {
diff --git a/weed-fs/src/pkg/topology/rack.go b/weed-fs/src/pkg/topology/rack.go
index f685f1a82..c819feb00 100644
--- a/weed-fs/src/pkg/topology/rack.go
+++ b/weed-fs/src/pkg/topology/rack.go
@@ -2,6 +2,7 @@ package topology
 
 import (
 	"strconv"
+	"time"
 )
 
 type Rack struct {
@@ -28,7 +29,8 @@ func (r *Rack) GetOrCreateDataNode(ip string, port int, publicUrl string, maxVol
 	for _, c := range r.Children() {
 		dn := c.(*DataNode)
 		if dn.MatchLocation(ip, port) {
-			dn.NodeImpl.UpAdjustMaxVolumeCountDelta(maxVolumeCount - dn.maxVolumeCount)
+			dn.LastSeen = time.Now().Unix()
+			dn.UpAdjustMaxVolumeCountDelta(maxVolumeCount - dn.maxVolumeCount)
 			return dn
 		}
 	}
@@ -37,6 +39,7 @@ func (r *Rack) GetOrCreateDataNode(ip string, port int, publicUrl string, maxVol
 	dn.Port = port
 	dn.PublicUrl = publicUrl
 	dn.maxVolumeCount = maxVolumeCount
+	dn.LastSeen = time.Now().Unix()
 	r.LinkChildNode(dn)
 	return dn
 }
diff --git a/weed-fs/src/pkg/topology/topology.go b/weed-fs/src/pkg/topology/topology.go
index 079fb54d9..a60768f14 100644
--- a/weed-fs/src/pkg/topology/topology.go
+++ b/weed-fs/src/pkg/topology/topology.go
@@ -6,6 +6,7 @@ import (
 	"pkg/directory"
 	"pkg/sequence"
 	"pkg/storage"
+	"time"
 )
 
 type Topology struct {
@@ -125,3 +126,19 @@ func (t *Topology) ToMap() interface{} {
 	m["layouts"] = layouts
 	return m
 }
+
+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
+}
diff --git a/weed-fs/src/pkg/topology/volume_location.go b/weed-fs/src/pkg/topology/volume_location.go
index 92d89ae46..f2e5dd894 100644
--- a/weed-fs/src/pkg/topology/volume_location.go
+++ b/weed-fs/src/pkg/topology/volume_location.go
@@ -27,7 +27,7 @@ func (dnll *DataNodeLocationList) Add(loc *DataNode) bool {
 func (dnll *DataNodeLocationList) Refresh(freshThreshHold int64) {
 	var changed bool
 	for _, dnl := range dnll.list {
-		if dnl.lastSeen < freshThreshHold {
+		if dnl.LastSeen < freshThreshHold {
 			changed = true
 			break
 		}
@@ -35,7 +35,7 @@ func (dnll *DataNodeLocationList) Refresh(freshThreshHold int64) {
 	if changed {
 		var l []*DataNode
 		for _, dnl := range dnll.list {
-			if dnl.lastSeen >= freshThreshHold {
+			if dnl.LastSeen >= freshThreshHold {
 				l = append(l, dnl)
 			}
 		}