tt: fix the race condition between start and stop

This patch fixes the signals handling strategy. Now there is only
one handling loop for all signals. Added a field for controlling
a state of the watchdog and a mutex for synchronizing the goroutines
changing this field.

Part of #325

Author: vr009

cli/running/running.go

@@ -6,7 +6,6 @@ import (
 	"io/ioutil"
 	"os"
 	"os/exec"
-	"os/signal"
 	"path"
 	"path/filepath"
 	"strings"
@@ -20,7 +19,6 @@ import (
 	"github.com/tarantool/tt/cli/process_utils"
 	"github.com/tarantool/tt/cli/ttlog"
 	"github.com/tarantool/tt/cli/util"
-	"golang.org/x/net/context"
 	"gopkg.in/yaml.v2"
 )
 
@@ -516,32 +514,21 @@ func FillCtx(cliOpts *config.CliOpts, cmdCtx *cmdcontext.CmdCtx,
 
 // Start an Instance.
 func Start(cmdCtx *cmdcontext.CmdCtx, run *InstanceCtx) error {
-	// Register a context, related with a signal handler that
-	// removes a pid file if interrupt signals were sent before starting
-	// an instance and watchdog signal handlers.
-	sigCtx, stop := signal.NotifyContext(context.Background(),
-		syscall.SIGINT, syscall.SIGTERM)
-	if err := process_utils.CreatePIDFile(run.PIDFile); err != nil {
-		return err
-	}
+	logger := createLogger(run)
+	provider := providerImpl{cmdCtx: cmdCtx, instanceCtx: run}
+	wd := NewWatchdog(run.Restartable, 5*time.Second, logger, &provider)
 
 	defer func() {
 		cleanup(run)
-		stop()
 	}()
 
-	logger := createLogger(run)
-	provider := providerImpl{cmdCtx: cmdCtx, instanceCtx: run}
-	wd := NewWatchdog(run.Restartable, 5*time.Second, logger, &provider)
-
-	// The Done() state of the context means that the process received
-	// an interrupt signal and there is a need to clean up the resources.
-	select {
-	case <-sigCtx.Done():
+	preStartAction := func() error {
+		if err := process_utils.CreatePIDFile(run.PIDFile); err != nil {
+			return err
+		}
 		return nil
-	default:
-		wd.Start()
 	}
+	wd.Start(preStartAction)
 	return nil
 }
 

cli/running/watchdog.go

@@ -35,17 +35,17 @@ type Watchdog struct {
 	// restartTimeout describes the timeout between
 	// restarting the Instance.
 	restartTimeout time.Duration
-	// stopped indicates whether Watchdog was stopped.
-	stopped bool
-	// stopMutex used to avoid a race condition under stopped field.
-	stopMutex sync.Mutex
 	// done channel used to inform the signal handle goroutine
 	// about termination of the Instance.
 	done chan bool
 	// provider provides Watchdog methods to get objects whose creation
 	// and updating may depend on changing external parameters
 	// (such as configuration file).
 	provider Provider
+	// stopMutex used to avoid a race condition under shouldStop field.
+	stopMutex sync.Mutex
+	// shouldStop indicates whether the Watchdog should be stopped.
+	shouldStop bool
 }
 
 // NewWatchdog creates a new instance of Watchdog.
@@ -55,33 +55,49 @@ func NewWatchdog(restartable bool, restartTimeout time.Duration, logger *ttlog.L
 		provider: provider}
 
 	wd.done = make(chan bool, 1)
-	wd.stopped = false
 
 	return &wd
 }
 
 // Start starts the Instance and signal handling.
-func (wd *Watchdog) Start() {
+func (wd *Watchdog) Start(preStartAction func() error) error {
+	var err error
+	// Create Instance.
+	if wd.Instance, err = wd.provider.CreateInstance(wd.logger); err != nil {
+		wd.logger.Printf(`Watchdog(ERROR): "%v".`, err)
+		return err
+	}
+	wd.logger = wd.Instance.logger
+	// The signal handling loop must be started before the instance
+	// get started for avoiding a race condition between tt start
+	// and tt stop. This way we avoid a situation when we receive
+	// a signal before starting a handler for it.
+	wd.startSignalHandling()
+
+	if err = preStartAction(); err != nil {
+		wd.logger.Printf(`Pre-start action error: %v`, err)
+		// Finish the signal handling goroutine.
+		wd.done <- true
+		return err
+	}
+
 	// The Instance must be restarted on completion if the "restartable"
 	// parameter is set to "true".
 	for {
 		var err error
-		// Create Instance.
-		if wd.Instance, err = wd.provider.CreateInstance(wd.logger); err != nil {
+
+		wd.stopMutex.Lock()
+		if wd.shouldStop {
+			wd.logger.Printf(`Watchdog(ERROR): terminated before instance start.`)
+			wd.stopMutex.Unlock()
+			return nil
+		}
+		// Start the Instance.
+		if err := wd.Instance.Start(); err != nil {
 			wd.logger.Printf(`Watchdog(ERROR): "%v".`, err)
+			wd.stopMutex.Unlock()
 			break
 		}
-		wd.logger = wd.Instance.logger
-		// Start the Instance and forwarding signals (except  SIGINT and SIGTERM)
-		wd.startSignalHandling()
-		wd.stopMutex.Lock()
-		if !wd.stopped {
-			if err := wd.Instance.Start(); err != nil {
-				wd.logger.Printf(`Watchdog(ERROR): "%v".`, err)
-				wd.stopMutex.Unlock()
-				break
-			}
-		}
 		wd.stopMutex.Unlock()
 
 		// Wait while the Instance will be terminated.
@@ -100,7 +116,7 @@ func (wd *Watchdog) Start() {
 			wd.logger.Println("Watchdog(ERROR): can't check if the instance is restartable.")
 			break
 		}
-		if wd.stopped || !restartable {
+		if wd.shouldStop || !restartable {
 			wd.logger.Println("Watchdog(INFO): the Instance has shutdown.")
 			break
 		}
@@ -112,15 +128,26 @@ func (wd *Watchdog) Start() {
 			wd.logger = logger
 		}
 		time.Sleep(wd.restartTimeout)
+
+		wd.shouldStop = false
+
+		// Recreate Instance.
+		if wd.Instance, err = wd.provider.CreateInstance(wd.logger); err != nil {
+			wd.logger.Printf(`Watchdog(ERROR): "%v".`, err)
+			return err
+		}
+		wd.logger = wd.Instance.logger
+		// Before the restart of an instance start a new signal handling loop.
+		wd.startSignalHandling()
 	}
+	return nil
 }
 
 // startSignalHandling starts signal handling in a separate goroutine.
 func (wd *Watchdog) startSignalHandling() {
 	sigChan := make(chan os.Signal, 1)
-	// Reset unregisters all previous handlers for interrupt signals.
-	signal.Reset(syscall.SIGINT,
-		syscall.SIGTERM, syscall.SIGHUP)
+	// Reset the signal mask before starting of the new loop.
+	signal.Reset()
 	signal.Notify(sigChan)
 
 	// Set barrier to synchronize with the main loop when the Instance stops.
@@ -137,19 +164,22 @@ func (wd *Watchdog) startSignalHandling() {
 				switch sig {
 				case syscall.SIGINT, syscall.SIGTERM:
 					wd.stopMutex.Lock()
-					wd.Instance.Stop(30 * time.Second)
 					// If we receive one of the "stop" signals, the
 					// program should be terminated.
-					wd.stopped = true
+					wd.shouldStop = true
 					wd.stopMutex.Unlock()
+					if wd.Instance.IsAlive() {
+						wd.Instance.Stop(30 * time.Second)
+					}
 				case syscall.SIGHUP:
 					// Rotate the log files.
 					wd.logger.Rotate()
 				default:
-					wd.Instance.SendSignal(sig)
+					if wd.Instance.IsAlive() {
+						wd.Instance.SendSignal(sig)
+					}
 				}
 			case _ = <-wd.done:
-				signal.Reset()
 				return
 			}
 		}

cli/running/watchdog_test.go

@@ -112,8 +112,9 @@ func TestWatchdogBase(t *testing.T) {
 	t.Cleanup(func() { cleanupWatchdog(wd) })
 
 	wdDoneChan := make(chan bool, 1)
+	testPreAction := func() error { return nil }
 	go func() {
-		wd.Start()
+		wd.Start(testPreAction)
 		wdDoneChan <- true
 	}()
 	waitProcessStart()
@@ -140,8 +141,9 @@ func TestWatchdogNotRestartable(t *testing.T) {
 	t.Cleanup(func() { cleanupWatchdog(wd) })
 
 	wdDoneChan := make(chan bool, 1)
+	testPreAction := func() error { return nil }
 	go func() {
-		wd.Start()
+		wd.Start(testPreAction)
 		wdDoneChan <- true
 	}()
 	waitProcessStart()