Merge pull request #79 from aszxqw/master

use sync.RWMutex when masterNode changes

go/weed/weed_server/volume_server.go

@@ -3,6 +3,7 @@ package weed_server
 import (
 	"math/rand"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/chrislusf/weed-fs/go/glog"
@@ -12,6 +13,7 @@ import (
 
 type VolumeServer struct {
 	masterNode   string
+	mnLock       sync.RWMutex
 	pulseSeconds int
 	dataCenter   string
 	rack         string
@@ -29,12 +31,12 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string,
 	whiteList []string,
 	fixJpgOrientation bool) *VolumeServer {
 	vs := &VolumeServer{
-		masterNode:        masterNode,
 		pulseSeconds:      pulseSeconds,
 		dataCenter:        dataCenter,
 		rack:              rack,
 		FixJpgOrientation: fixJpgOrientation,
 	}
+	vs.SetMasterNode(masterNode)
 	vs.store = storage.NewStore(port, adminPort, ip, publicUrl, folders, maxCounts)
 
 	vs.guard = security.NewGuard(whiteList, "")
@@ -54,7 +56,8 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string,
 
 	go func() {
 		connected := true
-		vs.store.SetBootstrapMaster(vs.masterNode)
+
+		vs.store.SetBootstrapMaster(vs.GetMasterNode())
 		vs.store.SetDataCenter(vs.dataCenter)
 		vs.store.SetRack(vs.rack)
 		for {
@@ -62,8 +65,8 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string,
 			if err == nil {
 				if !connected {
 					connected = true
-					vs.masterNode = master
+					vs.SetMasterNode(master)
-					glog.V(0).Infoln("Volume Server Connected with master at", master)
+					glog.V(0).Infoln("Volume Server Connected with master at", master, "and set it as masterNode")
 				}
 			} else {
 				glog.V(4).Infoln("Volume Server Failed to talk with master:", err.Error())
@@ -82,6 +85,18 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string,
 	return vs
 }
 
+func (vs *VolumeServer) GetMasterNode() string {
+	vs.mnLock.RLock()
+	defer vs.mnLock.RUnlock()
+	return vs.masterNode
+}
+
+func (vs *VolumeServer) SetMasterNode(masterNode string) {
+	vs.mnLock.Lock()
+	defer vs.mnLock.Unlock()
+	vs.masterNode = masterNode
+}
+
 func (vs *VolumeServer) Shutdown() {
 	glog.V(0).Infoln("Shutting down volume server...")
 	vs.store.Close()

go/weed/weed_server/volume_server_handlers.go

@@ -58,7 +58,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 
 	glog.V(4).Infoln("volume", volumeId, "reading", n)
 	if !vs.store.HasVolume(volumeId) {
-		lookupResult, err := operation.Lookup(vs.masterNode, volumeId.String())
+		lookupResult, err := operation.Lookup(vs.GetMasterNode(), volumeId.String())
 		glog.V(2).Infoln("volume", volumeId, "found on", lookupResult, "error", err)
 		if err == nil && len(lookupResult.Locations) > 0 {
 			http.Redirect(w, r, "http://"+lookupResult.Locations[0].Url+r.URL.Path, http.StatusMovedPermanently)
@@ -253,7 +253,7 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	ret := operation.UploadResult{}
-	size, errorStatus := topology.ReplicatedWrite(vs.masterNode, vs.store, volumeId, needle, r)
+	size, errorStatus := topology.ReplicatedWrite(vs.GetMasterNode(), vs.store, volumeId, needle, r)
 	httpStatus := http.StatusCreated
 	if errorStatus != "" {
 		httpStatus = http.StatusInternalServerError
@@ -290,7 +290,7 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	n.Size = 0
-	ret := topology.ReplicatedDelete(vs.masterNode, vs.store, volumeId, n, r)
+	ret := topology.ReplicatedDelete(vs.GetMasterNode(), vs.store, volumeId, n, r)
 
 	if ret != 0 {
 		m := make(map[string]uint32)