Merge pull request #2940 from yichengq/improve-raft-loop

etcdserver: stop raft loop when receiving stop signal

etcdserver/raft.go

@@ -161,7 +161,11 @@ func (r *raftNode) run() {
 
 			r.s.send(rd.Messages)
 
-			<-apply.done
+			select {
+			case <-apply.done:
+			case <-r.stopped:
+				return
+			}
 			r.Advance()
 		case <-syncC:
 			r.s.sync(defaultSyncTimeout)

etcdserver/raft_test.go

@@ -18,9 +18,11 @@ import (
 	"encoding/json"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/coreos/etcd/pkg/pbutil"
 	"github.com/coreos/etcd/pkg/types"
+	"github.com/coreos/etcd/raft"
 	"github.com/coreos/etcd/raft/raftpb"
 )
 
@@ -141,3 +143,29 @@ func TestCreateConfigChangeEnts(t *testing.T) {
 		}
 	}
 }
+
+func TestStopRaftWhenWaitingForApplyDone(t *testing.T) {
+	n := newReadyNode()
+	r := raftNode{
+		Node:        n,
+		applyc:      make(chan apply),
+		storage:     &storageRecorder{},
+		raftStorage: raft.NewMemoryStorage(),
+		transport:   &nopTransporter{},
+	}
+	r.s = &EtcdServer{r: r}
+	go r.run()
+	n.readyc <- raft.Ready{}
+	select {
+	case <-r.applyc:
+	case <-time.After(time.Second):
+		t.Fatalf("failed to receive apply struct")
+	}
+
+	r.stopped <- struct{}{}
+	select {
+	case <-r.done:
+	case <-time.After(time.Second):
+		t.Fatalf("failed to stop raft loop")
+	}
+}