Merge pull request #575 from xiaost/skip-persist-meta-while-loading

nsqd: skip persist metadata while loading

nsqd/channel.go

@@ -296,12 +296,7 @@ func (c *Channel) doPause(pause bool) error {
 		}
 	}
 	c.RUnlock()
-
-	c.ctx.nsqd.Lock()
-	defer c.ctx.nsqd.Unlock()
-	// pro-actively persist metadata so in case of process failure
-	// nsqd won't suddenly (un)pause a channel
-	return c.ctx.nsqd.PersistMetadata()
+	return nil
 }
 
 func (c *Channel) IsPaused() bool {

nsqd/http.go

@@ -415,6 +415,11 @@ func (s *httpServer) doPauseTopic(req *http.Request) (interface{}, error) {
 		return nil, http_api.Err{500, "INTERNAL_ERROR"}
 	}
 
+	// pro-actively persist metadata so in case of process failure
+	// nsqd won't suddenly (un)pause a topic
+	s.ctx.nsqd.Lock()
+	s.ctx.nsqd.PersistMetadata()
+	s.ctx.nsqd.Unlock()
 	return nil, nil
 }
 
@@ -481,6 +486,11 @@ func (s *httpServer) doPauseChannel(req *http.Request) (interface{}, error) {
 		return nil, http_api.Err{500, "INTERNAL_ERROR"}
 	}
 
+	// pro-actively persist metadata so in case of process failure
+	// nsqd won't suddenly (un)pause a channel
+	s.ctx.nsqd.Lock()
+	s.ctx.nsqd.PersistMetadata()
+	s.ctx.nsqd.Unlock()
 	return nil, nil
 }
 

nsqd/nsqd.go

@@ -32,6 +32,11 @@ const (
 	TLSRequired
 )
 
+const (
+	flagHealthy = 1 << iota
+	flagLoading
+)
+
 type NSQD struct {
 	// 64bit atomic vars need to be first for proper alignment on 32bit platforms
 	clientIDSequence int64
@@ -40,8 +45,8 @@ type NSQD struct {
 
 	opts *nsqdOptions
 
-	healthMtx sync.RWMutex
-	healthy   int32
+	flag      int32
+	errMtx    sync.RWMutex
 	err       error
 	startTime time.Time
 
@@ -63,7 +68,7 @@ type NSQD struct {
 func NewNSQD(opts *nsqdOptions) *NSQD {
 	n := &NSQD{
 		opts:       opts,
-		healthy:    1,
+		flag:       flagHealthy,
 		startTime:  time.Now(),
 		topicMap:   make(map[string]*Topic),
 		idChan:     make(chan MessageID, 4096),
@@ -135,24 +140,43 @@ func (n *NSQD) RealHTTPAddr() *net.TCPAddr {
 	return n.httpListener.Addr().(*net.TCPAddr)
 }
 
+func (n *NSQD) setFlag(f int32, b bool) {
+	for {
+		old := atomic.LoadInt32(&n.flag)
+		newFlag := old
+		if b {
+			newFlag |= f
+		} else {
+			newFlag &= ^f
+		}
+		if atomic.CompareAndSwapInt32(&n.flag, old, newFlag) {
+			return
+		}
+	}
+}
+
+func (n *NSQD) getFlag(f int32) bool {
+	return f&atomic.LoadInt32(&n.flag) != 0
+}
+
 func (n *NSQD) SetHealth(err error) {
-	n.healthMtx.Lock()
-	defer n.healthMtx.Unlock()
+	n.errMtx.Lock()
+	defer n.errMtx.Unlock()
 	n.err = err
 	if err != nil {
-		atomic.StoreInt32(&n.healthy, 0)
+		n.setFlag(flagHealthy, false)
 	} else {
-		atomic.StoreInt32(&n.healthy, 1)
+		n.setFlag(flagHealthy, true)
 	}
 }
 
 func (n *NSQD) IsHealthy() bool {
-	return atomic.LoadInt32(&n.healthy) == 1
+	return n.getFlag(flagHealthy)
 }
 
 func (n *NSQD) GetError() error {
-	n.healthMtx.RLock()
-	defer n.healthMtx.RUnlock()
+	n.errMtx.RLock()
+	defer n.errMtx.RUnlock()
 	return n.err
 }
 
@@ -229,6 +253,8 @@ func (n *NSQD) Main() {
 }
 
 func (n *NSQD) LoadMetadata() {
+	n.setFlag(flagLoading, true)
+	defer n.setFlag(flagLoading, false)
 	fn := fmt.Sprintf(path.Join(n.opts.DataPath, "nsqd.%d.dat"), n.opts.ID)
 	data, err := ioutil.ReadFile(fn)
 	if err != nil {
@@ -502,12 +528,19 @@ exit:
 }
 
 func (n *NSQD) Notify(v interface{}) {
+	// since the in-memory metadata is incomplete,
+	// should not persist metadata while loading it.
+	// nsqd will call `PersistMetadata` it after loading
+	persist := !n.getFlag(flagLoading)
 	n.waitGroup.Wrap(func() {
 		// by selecting on exitChan we guarantee that
 		// we do not block exit, see issue #123
 		select {
 		case <-n.exitChan:
 		case n.notifyChan <- v:
+			if !persist {
+				return
+			}
 			n.Lock()
 			err := n.PersistMetadata()
 			if err != nil {

nsqd/nsqd_test.go

@@ -94,11 +94,15 @@ func TestStartup(t *testing.T) {
 	// verify nsqd metadata shows no topics
 	err := nsqd.PersistMetadata()
 	equal(t, err, nil)
+	nsqd.setFlag(flagLoading, true)
+	nsqd.GetTopic(topicName) // will not persist if `flagLoading`
 	metaData, err := getMetadata(nsqd)
 	equal(t, err, nil)
 	topics, err := metaData.Get("topics").Array()
 	equal(t, err, nil)
 	equal(t, len(topics), 0)
+	nsqd.DeleteExistingTopic(topicName)
+	nsqd.setFlag(flagLoading, false)
 
 	body := make([]byte, 256)
 	topic := nsqd.GetTopic(topicName)
@@ -246,12 +250,18 @@ func TestPauseMetadata(t *testing.T) {
 	equal(t, b, false)
 
 	channel.Pause()
+	b, _ = metadataForChannel(nsqd, 0, 0).Get("paused").Bool()
+	equal(t, b, false)
 
+	nsqd.PersistMetadata()
 	b, _ = metadataForChannel(nsqd, 0, 0).Get("paused").Bool()
 	equal(t, b, true)
 
 	channel.UnPause()
+	b, _ = metadataForChannel(nsqd, 0, 0).Get("paused").Bool()
+	equal(t, b, true)
 
+	nsqd.PersistMetadata()
 	b, _ = metadataForChannel(nsqd, 0, 0).Get("paused").Bool()
 	equal(t, b, false)
 }

nsqd/topic.go

@@ -413,11 +413,6 @@ func (t *Topic) doPause(pause bool) error {
 
 	select {
 	case t.pauseChan <- pause:
-		t.ctx.nsqd.Lock()
-		defer t.ctx.nsqd.Unlock()
-		// pro-actively persist metadata so in case of process failure
-		// nsqd won't suddenly (un)pause a topic
-		return t.ctx.nsqd.PersistMetadata()
 	case <-t.exitChan:
 	}