refactor: clean up duplicate server code

Signed-off-by: Shuoran Liu <shuoranliu@gmail.com>

cmd/cmd.go

@@ -33,6 +33,7 @@ import (
 	"github.com/jacobsa/daemonize"
 
 	"github.com/chubaofs/chubaofs/authnode"
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/datanode"
 	"github.com/chubaofs/chubaofs/master"
 	"github.com/chubaofs/chubaofs/metanode"
@@ -81,14 +82,7 @@ var (
 	configForeground = flag.Bool("f", false, "run foreground")
 )
 
-type Server interface {
-	Start(cfg *config.Config) error
-	Shutdown()
-	// Sync will block invoker goroutine until this MetaNode shutdown.
-	Sync()
-}
-
-func interceptSignal(s Server) {
+func interceptSignal(s common.Server) {
 	sigC := make(chan os.Signal, 1)
 	signal.Notify(sigC, syscall.SIGINT, syscall.SIGTERM)
 	syslog.Println("action[interceptSignal] register system signal.")
@@ -161,7 +155,7 @@ func main() {
 
 	// Init server instance with specified role configuration.
 	var (
-		server Server
+		server common.Server
 		module string
 	)
 	switch role {

cmd/common/server.go

@@ -0,0 +1,66 @@
+package common
+
+import (
+	"sync"
+	"sync/atomic"
+
+	"github.com/chubaofs/chubaofs/util/config"
+)
+
+const (
+	StateStandby uint32 = iota
+	StateStart
+	StateRunning
+	StateShutdown
+	StateStopped
+)
+
+type Control struct {
+	state uint32
+	wg    sync.WaitGroup
+}
+
+type Server interface {
+	Start(cfg *config.Config) error
+	Shutdown()
+	// Sync will block invoker goroutine until this MetaNode shutdown.
+	Sync()
+}
+
+type DoStartFunc func(s Server, cfg *config.Config) (err error)
+type DoShutdownFunc func(s Server)
+
+func (c *Control) Start(s Server, cfg *config.Config, do DoStartFunc) (err error) {
+	if atomic.CompareAndSwapUint32(&c.state, StateStandby, StateStart) {
+		defer func() {
+			var newState uint32
+			if err != nil {
+				newState = StateStandby
+			} else {
+				newState = StateRunning
+			}
+			atomic.StoreUint32(&c.state, newState)
+		}()
+		if err = do(s, cfg); err != nil {
+			return
+		}
+		c.wg.Add(1)
+	}
+	return
+
+}
+
+func (c *Control) Shutdown(s Server, do DoShutdownFunc) {
+	if atomic.CompareAndSwapUint32(&c.state, StateRunning, StateShutdown) {
+		do(s)
+		c.wg.Done()
+		atomic.StoreUint32(&c.state, StateStopped)
+	}
+
+}
+
+func (c *Control) Sync() {
+	if atomic.LoadUint32(&c.state) == StateRunning {
+		c.wg.Wait()
+	}
+}

datanode/const.go

@@ -14,15 +14,6 @@
 
 package datanode
 
-// The status of the server
-const (
-	Standby uint32 = iota
-	Start
-	Running
-	Shutdown
-	Stopped
-)
-
 const (
 	IntervalToUpdateReplica       = 600 // interval to update the replica
 	IntervalToUpdatePartitionSize = 60  // interval to update the partition size

datanode/server.go

@@ -23,13 +23,13 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"errors"
 	"os"
 	"syscall"
 
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/proto"
 	"github.com/chubaofs/chubaofs/raftstore"
 	"github.com/chubaofs/chubaofs/repl"
@@ -45,9 +45,9 @@ var (
 	ErrNoSpaceToCreatePartition = errors.New("No disk space to create a data partition")
 	ErrNewSpaceManagerFailed    = errors.New("Creater new space manager failed")
 
-	LocalIP ,serverPort     string
-	gConnPool    = util.NewConnectPool()
-	MasterClient = masterSDK.NewMasterClient(nil, false)
+	LocalIP, serverPort string
+	gConnPool           = util.NewConnectPool()
+	MasterClient        = masterSDK.NewMasterClient(nil, false)
 )
 
 const (
@@ -90,8 +90,8 @@ type DataNode struct {
 
 	tcpListener net.Listener
 	stopC       chan bool
-	state       uint32
-	wg          sync.WaitGroup
+
+	control common.Control
 }
 
 func NewServer() *DataNode {
@@ -100,40 +100,26 @@ func NewServer() *DataNode {
 
 func (s *DataNode) Start(cfg *config.Config) (err error) {
 	runtime.GOMAXPROCS(runtime.NumCPU())
-	if atomic.CompareAndSwapUint32(&s.state, Standby, Start) {
-		defer func() {
-			if err != nil {
-				atomic.StoreUint32(&s.state, Standby)
-			} else {
-				atomic.StoreUint32(&s.state, Running)
-			}
-		}()
-		if err = s.onStart(cfg); err != nil {
-			return
-		}
-		s.wg.Add(1)
-	}
-	return
+	return s.control.Start(s, cfg, doStart)
 }
 
 // Shutdown shuts down the current data node.
 func (s *DataNode) Shutdown() {
-	if atomic.CompareAndSwapUint32(&s.state, Running, Shutdown) {
-		s.onShutdown()
-		s.wg.Done()
-		atomic.StoreUint32(&s.state, Stopped)
-	}
+	s.control.Shutdown(s, doShutdown)
 }
 
 // Sync keeps data node in sync.
 func (s *DataNode) Sync() {
-	if atomic.LoadUint32(&s.state) == Running {
-		s.wg.Wait()
-	}
+	s.control.Sync()
 }
 
 // Workflow of starting up a data node.
-func (s *DataNode) onStart(cfg *config.Config) (err error) {
+func doStart(server common.Server, cfg *config.Config) (err error) {
+	s, ok := server.(*DataNode)
+	if !ok {
+		return errors.New("Invalid Node Type!")
+	}
+
 	s.stopC = make(chan bool, 0)
 
 	// parse the config file
@@ -170,11 +156,14 @@ func (s *DataNode) onStart(cfg *config.Config) (err error) {
 	return
 }
 
-func (s *DataNode) onShutdown() {
+func doShutdown(server common.Server) {
+	s, ok := server.(*DataNode)
+	if !ok {
+		return
+	}
 	close(s.stopC)
 	s.stopTCPService()
 	s.stopRaftServer()
-	return
 }
 
 func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
@@ -184,7 +173,7 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
 	)
 	LocalIP = cfg.GetString(ConfigKeyLocalIP)
 	port = cfg.GetString(proto.ListenPort)
-	serverPort=port
+	serverPort = port
 	if regexpPort, err = regexp.Compile("^(\\d)+$"); err != nil {
 		return fmt.Errorf("Err:no port")
 	}

metanode/const.go

@@ -20,14 +20,6 @@ import (
 	"time"
 )
 
-const (
-	StateStandby uint32 = iota
-	StateStart
-	StateRunning
-	StateShutdown
-	StateStopped
-)
-
 // Type alias.
 type (
 	// Master -> MetaNode  create metaPartition request

metanode/manager.go

@@ -27,6 +27,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/proto"
 	"github.com/chubaofs/chubaofs/raftstore"
 	"github.com/chubaofs/chubaofs/util"
@@ -161,13 +162,13 @@ func (m *metadataManager) HandleMetadataOperation(conn net.Conn, p *Packet,
 
 // Start starts the metadata manager.
 func (m *metadataManager) Start() (err error) {
-	if atomic.CompareAndSwapUint32(&m.state, StateStandby, StateStart) {
+	if atomic.CompareAndSwapUint32(&m.state, common.StateStandby, common.StateStart) {
 		defer func() {
 			var newState uint32
 			if err != nil {
-				newState = StateStandby
+				newState = common.StateStandby
 			} else {
-				newState = StateRunning
+				newState = common.StateRunning
 			}
 			atomic.StoreUint32(&m.state, newState)
 		}()
@@ -178,8 +179,8 @@ func (m *metadataManager) Start() (err error) {
 
 // Stop stops the metadata manager.
 func (m *metadataManager) Stop() {
-	if atomic.CompareAndSwapUint32(&m.state, StateRunning, StateShutdown) {
-		defer atomic.StoreUint32(&m.state, StateStopped)
+	if atomic.CompareAndSwapUint32(&m.state, common.StateRunning, common.StateShutdown) {
+		defer atomic.StoreUint32(&m.state, common.StateStopped)
 		m.onStop()
 	}
 }

metanode/metanode.go

@@ -17,8 +17,6 @@ package metanode
 import (
 	"os"
 	"strings"
-	"sync"
-	"sync/atomic"
 	"time"
 
 	masterSDK "github.com/chubaofs/chubaofs/sdk/master"
@@ -26,6 +24,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/proto"
 	"github.com/chubaofs/chubaofs/raftstore"
 	"github.com/chubaofs/chubaofs/util"
@@ -39,7 +38,7 @@ var (
 	clusterInfo    *proto.ClusterInfo
 	masterClient   *masterSDK.MasterClient
 	configTotalMem uint64
-	serverPort       string
+	serverPort     string
 )
 
 // The MetaNode manages the dentry and inode information of the meta partitions on a meta node.
@@ -56,8 +55,8 @@ type MetaNode struct {
 	raftHeartbeatPort string
 	raftReplicatePort string
 	httpStopC         chan uint8
-	state             uint32
-	wg                sync.WaitGroup
+
+	control common.Control
 }
 
 // Start starts up the meta node with the specified configuration.
@@ -65,31 +64,12 @@ type MetaNode struct {
 //  2. Restore raftStore fsm of each meta node range.
 //  3. Start server and accept connection from the master and clients.
 func (m *MetaNode) Start(cfg *config.Config) (err error) {
-	if atomic.CompareAndSwapUint32(&m.state, StateStandby, StateStart) {
-		defer func() {
-			var newState uint32
-			if err != nil {
-				newState = StateStandby
-			} else {
-				newState = StateRunning
-			}
-			atomic.StoreUint32(&m.state, newState)
-		}()
-		if err = m.onStart(cfg); err != nil {
-			return
-		}
-		m.wg.Add(1)
-	}
-	return
+	return m.control.Start(m, cfg, doStart)
 }
 
 // Shutdown stops the meta node.
 func (m *MetaNode) Shutdown() {
-	if atomic.CompareAndSwapUint32(&m.state, StateRunning, StateShutdown) {
-		defer atomic.StoreUint32(&m.state, StateStopped)
-		m.onShutdown()
-		m.wg.Done()
-	}
+	m.control.Shutdown(m, doShutdown)
 }
 
 func (m *MetaNode) checkLocalPartitionMatchWithMaster() (err error) {
@@ -120,7 +100,11 @@ func (m *MetaNode) checkLocalPartitionMatchWithMaster() (err error) {
 	return
 }
 
-func (m *MetaNode) onStart(cfg *config.Config) (err error) {
+func doStart(s common.Server, cfg *config.Config) (err error) {
+	m, ok := s.(*MetaNode)
+	if !ok {
+		return errors.New("Invalid Node Type!")
+	}
 	if err = m.parseConfig(cfg); err != nil {
 		return
 	}
@@ -153,7 +137,11 @@ func (m *MetaNode) onStart(cfg *config.Config) (err error) {
 	return
 }
 
-func (m *MetaNode) onShutdown() {
+func doShutdown(s common.Server) {
+	m, ok := s.(*MetaNode)
+	if !ok {
+		return
+	}
 	// shutdown node and release the resource
 	m.stopServer()
 	m.stopMetaManager()
@@ -162,9 +150,7 @@ func (m *MetaNode) onShutdown() {
 
 // Sync blocks the invoker's goroutine until the meta node shuts down.
 func (m *MetaNode) Sync() {
-	if atomic.LoadUint32(&m.state) == StateRunning {
-		m.wg.Wait()
-	}
+	m.control.Sync()
 }
 
 func (m *MetaNode) parseConfig(cfg *config.Config) (err error) {
@@ -174,7 +160,7 @@ func (m *MetaNode) parseConfig(cfg *config.Config) (err error) {
 	}
 	m.localAddr = cfg.GetString(cfgLocalIP)
 	m.listen = cfg.GetString(proto.ListenPort)
-	serverPort=m.listen
+	serverPort = m.listen
 	m.metadataDir = cfg.GetString(cfgMetadataDir)
 	m.raftDir = cfg.GetString(cfgRaftDir)
 	m.raftHeartbeatPort = cfg.GetString(cfgRaftHeartbeatPort)

metanode/partition.go

@@ -27,6 +27,7 @@ import (
 	"os"
 	"path"
 
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/proto"
 	"github.com/chubaofs/chubaofs/raftstore"
 	"github.com/chubaofs/chubaofs/util"
@@ -212,13 +213,13 @@ type metaPartition struct {
 
 // Start starts a meta partition.
 func (mp *metaPartition) Start() (err error) {
-	if atomic.CompareAndSwapUint32(&mp.state, StateStandby, StateStart) {
+	if atomic.CompareAndSwapUint32(&mp.state, common.StateStandby, common.StateStart) {
 		defer func() {
 			var newState uint32
 			if err != nil {
-				newState = StateStandby
+				newState = common.StateStandby
 			} else {
-				newState = StateRunning
+				newState = common.StateRunning
 			}
 			atomic.StoreUint32(&mp.state, newState)
 		}()
@@ -238,8 +239,8 @@ func (mp *metaPartition) Start() (err error) {
 
 // Stop stops a meta partition.
 func (mp *metaPartition) Stop() {
-	if atomic.CompareAndSwapUint32(&mp.state, StateRunning, StateShutdown) {
-		defer atomic.StoreUint32(&mp.state, StateStopped)
+	if atomic.CompareAndSwapUint32(&mp.state, common.StateRunning, common.StateShutdown) {
+		defer atomic.StoreUint32(&mp.state, common.StateStopped)
 		if mp.config.BeforeStop != nil {
 			mp.config.BeforeStop()
 		}

metanode/partition_store_ticket.go

@@ -17,6 +17,7 @@ package metanode
 import (
 	"time"
 
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/util/errors"
 	"github.com/chubaofs/chubaofs/util/exporter"
 	"github.com/chubaofs/chubaofs/util/log"
@@ -34,7 +35,7 @@ type storeMsg struct {
 func (mp *metaPartition) startSchedule(curIndex uint64) {
 	timer := time.NewTimer(time.Hour * 24 * 365)
 	timer.Stop()
-	scheduleState := StateStopped
+	scheduleState := common.StateStopped
 	dumpFunc := func(msg *storeMsg) {
 		log.LogDebugf("[startSchedule] partitionId=%d: nowAppID"+
 			"=%d, applyID=%d", mp.config.PartitionId, curIndex,
@@ -61,15 +62,15 @@ func (mp *metaPartition) startSchedule(curIndex uint64) {
 		if _, ok := mp.IsLeader(); ok {
 			timer.Reset(intervalToPersistData)
 		}
-		scheduleState = StateStopped
+		scheduleState = common.StateStopped
 	}
 	go func(stopC chan bool) {
 		var msgs []*storeMsg
 		readyChan := make(chan struct{}, 1)
 		for {
 			if len(msgs) > 0 {
-				if scheduleState == StateStopped {
-					scheduleState = StateRunning
+				if scheduleState == common.StateStopped {
+					scheduleState = common.StateRunning
 					readyChan <- struct{}{}
 				}
 			}

objectnode/server.go

@@ -19,24 +19,14 @@ import (
 	"github.com/chubaofs/chubaofs/proto"
 	"net/http"
 	"regexp"
-	"sync"
-	"sync/atomic"
 
+	"github.com/chubaofs/chubaofs/cmd/common"
 	"github.com/chubaofs/chubaofs/util/config"
 	"github.com/chubaofs/chubaofs/util/errors"
 	"github.com/chubaofs/chubaofs/util/log"
 	"github.com/gorilla/mux"
 )
 
-// The status of the s3 server
-const (
-	Standby uint32 = iota
-	Start
-	Running
-	Shutdown
-	Stopped
-)
-
 // Configuration keys
 const (
 	configListen    = "listen"
@@ -63,39 +53,20 @@ type ObjectNode struct {
 	region     string
 	httpServer *http.Server
 	vm         VolumeManager
-	state      uint32
-	wg         sync.WaitGroup
+
+	control common.Control
 }
 
 func (o *ObjectNode) Start(cfg *config.Config) (err error) {
-	if atomic.CompareAndSwapUint32(&o.state, Standby, Start) {
-		defer func() {
-			if err != nil {
-				atomic.StoreUint32(&o.state, Standby)
-			} else {
-				atomic.StoreUint32(&o.state, Running)
-			}
-		}()
-		if err = o.handleStart(cfg); err != nil {
-			return
-		}
-		o.wg.Add(1)
-	}
-	return
+	return o.control.Start(o, cfg, handleStart)
 }
 
 func (o *ObjectNode) Shutdown() {
-	if atomic.CompareAndSwapUint32(&o.state, Running, Shutdown) {
-		o.handleShutdown()
-		o.wg.Done()
-		atomic.StoreUint32(&o.state, Stopped)
-	}
+	o.control.Shutdown(o, handleShutdown)
 }
 
 func (o *ObjectNode) Sync() {
-	if atomic.LoadUint32(&o.state) == Running {
-		o.wg.Wait()
-	}
+	o.control.Sync()
 }
 
 func (o *ObjectNode) parseConfig(cfg *config.Config) (err error) {
@@ -139,7 +110,11 @@ func (o *ObjectNode) parseConfig(cfg *config.Config) (err error) {
 	return
 }
 
-func (o *ObjectNode) handleStart(cfg *config.Config) (err error) {
+func handleStart(s common.Server, cfg *config.Config) (err error) {
+	o, ok := s.(*ObjectNode)
+	if !ok {
+		return errors.New("Invalid Node Type!")
+	}
 	// parse config
 	if err = o.parseConfig(cfg); err != nil {
 		return
@@ -153,7 +128,11 @@ func (o *ObjectNode) handleStart(cfg *config.Config) (err error) {
 	return
 }
 
-func (o *ObjectNode) handleShutdown() {
+func handleShutdown(s common.Server) {
+	o, ok := s.(*ObjectNode)
+	if !ok {
+		return
+	}
 	o.shutdownRestAPI()
 }