Issue 11: Failed to write to replicas for volumen 3

Avoid unnecessary master lookup

2 files changed, 20 insertions, 9 deletions

diff --git a/weed-fs/src/cmd/weed/volume.go b/weed-fs/src/cmd/weed/volume.go
index 9e6f3e9ac..2ff84f16b 100644
--- a/weed-fs/src/cmd/weed/volume.go
+++ b/weed-fs/src/cmd/weed/volume.go
@@ -67,13 +67,13 @@ func vacuumVolumeCompactHandler(w http.ResponseWriter, r *http.Request) {
 	debug("compacted volume =", r.FormValue("volume"), ", error =", err)
 }
 func vacuumVolumeCommitHandler(w http.ResponseWriter, r *http.Request) {
-  count, err := store.CommitCompactVolume(r.FormValue("volume"))
-  if err == nil {
-    writeJson(w, r, map[string]interface{}{"error": "", "size":count})
-  } else {
-    writeJson(w, r, map[string]string{"error": err.Error()})
-  }
-  debug("commit compact volume =", r.FormValue("volume"), ", error =", err)
+	count, err := store.CommitCompactVolume(r.FormValue("volume"))
+	if err == nil {
+		writeJson(w, r, map[string]interface{}{"error": "", "size": count})
+	} else {
+		writeJson(w, r, map[string]string{"error": err.Error()})
+	}
+	debug("commit compact volume =", r.FormValue("volume"), ", error =", err)
 }
 func storeHandler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
@@ -146,7 +146,11 @@ func PostHandler(w http.ResponseWriter, r *http.Request) {
 		} else {
 			ret := store.Write(volumeId, needle)
 			errorStatus := ""
-			if ret > 0 || !store.HasVolume(volumeId) { //send to other replica locations
+			needToReplicate := !store.HasVolume(volumeId)
+			if !needToReplicate && ret > 0 {
+				needToReplicate = store.GetVolume(volumeId).NeedToReplicate()
+			}
+			if needToReplicate { //send to other replica locations
 				if r.FormValue("type") != "standard" {
 					if !distributedOperation(volumeId, func(location operation.Location) bool {
 						_, err := operation.Upload("http://"+location.Url+r.URL.Path+"?type=standard", filename, bytes.NewReader(needle.Data))
@@ -201,7 +205,11 @@ func DeleteHandler(w http.ResponseWriter, r *http.Request) {
 	n.Size = 0
 	ret := store.Delete(volumeId, n)
 
-	if ret > 0 || !store.HasVolume(volumeId) { //send to other replica locations
+	needToReplicate := !store.HasVolume(volumeId)
+	if !needToReplicate && ret > 0 {
+		needToReplicate = store.GetVolume(volumeId).NeedToReplicate()
+	}
+	if needToReplicate { //send to other replica locations
 		if r.FormValue("type") != "standard" {
 			if !distributedOperation(volumeId, func(location operation.Location) bool {
 				return nil == operation.Delete("http://"+location.Url+r.URL.Path+"?type=standard")
diff --git a/weed-fs/src/pkg/storage/volume.go b/weed-fs/src/pkg/storage/volume.go
index ee1d98b6a..3bc979251 100644
--- a/weed-fs/src/pkg/storage/volume.go
+++ b/weed-fs/src/pkg/storage/volume.go
@@ -73,6 +73,9 @@ func (v *Volume) readSuperBlock() {
 		v.replicaType, _ = NewReplicationTypeFromByte(header[1])
 	}
 }
+func (v *Volume) NeedToReplicate() bool{
+  return v.replicaType.GetCopyCount()>1
+}
 
 func (v *Volume) write(n *Needle) uint32 {
 	v.accessLock.Lock()