carbon/grace.go
func (app *App) DumpStop() error {
	app.Lock()
	defer app.Unlock()

	if !app.Config.Dump.Enabled {
		return nil
	}

	if app.Persister != nil {
		app.Persister.Stop()
		app.Persister = nil
	}

	logger := zapwriter.Logger("dump")

	logger.Info("grace stop with dump inited")

	filenamePostfix := fmt.Sprintf("%d.%d", os.Getpid(), time.Now().UnixNano())
	dumpFilename := path.Join(app.Config.Dump.Path, fmt.Sprintf("cache.%s.bin", filenamePostfix))
	xlogFilename := path.Join(app.Config.Dump.Path, fmt.Sprintf("input.%s", filenamePostfix))

	// start dumpers
	logger.Info("start cache dump", zap.String("filename", dumpFilename))
	logger.Info("start wal write", zap.String("filename", xlogFilename))

	// open dump file
	dump, err := os.Create(dumpFilename)
	if err != nil {
		return err
	}
	dumpWriter := bufio.NewWriterSize(dump, 1048576) // 1Mb

	// start input dumper
	xlog, err := os.Create(xlogFilename)
	if err != nil {
		return err
	}
	xlogWriter := &SyncWriter{w: bufio.NewWriterSize(xlog, 4096)} // 4kb

	app.Cache.DivertToXlog(xlogWriter)

	// stop cache
	dumpStart := time.Now()
	cacheSize := app.Cache.Size()

	// dump cache
	err = app.Cache.DumpBinary(dumpWriter)
	if err != nil {
		logger.Info("dump failed", zap.Error(err))
		return err
	}

	logger.Info("cache dump finished",
		zap.Int("records", int(cacheSize)),
		zap.Duration("runtime", time.Since(dumpStart)),
	)

	if err = dumpWriter.Flush(); err != nil {
		logger.Info("dump flush failed", zap.Error(err))
		return err
	}

	if err = dump.Close(); err != nil {
		logger.Info("dump close failed", zap.Error(err))
		return err
	}

	// cache dump finished

	logger.Info("dump finished")

	logger.Info("stop listeners")

	stopped := make(chan struct{})

	go func() {
		app.stopListeners()

		if err := xlogWriter.Flush(); err != nil {
			logger.Info("xlog flush failed", zap.Error(err))
			return
		}

		if err := xlog.Close(); err != nil {
			logger.Info("xlog close failed", zap.Error(err))
			return
		}

		close(stopped)
	}()

	select {
	case <-time.After(5 * time.Second):
		logger.Info("stop listeners timeout, force stop daemon")
	case <-stopped:
		logger.Info("listeners stopped")
	}

	// logger.Info("stop all")
	// app.stopAll()

	return nil
}
func (app *App) DumpStop() error {
	app.Lock()
	defer app.Unlock()

	if !app.Config.Dump.Enabled {
		return nil
	}

	if app.Persister != nil {
		app.Persister.Stop()
		app.Persister = nil
	}

	logger := zapwriter.Logger("dump")

	logger.Info("grace stop with dump inited")

	filenamePostfix := fmt.Sprintf("%d.%d", os.Getpid(), time.Now().UnixNano())
	dumpFilename := path.Join(app.Config.Dump.Path, fmt.Sprintf("cache.%s.bin", filenamePostfix))
	xlogFilename := path.Join(app.Config.Dump.Path, fmt.Sprintf("input.%s", filenamePostfix))

	// start dumpers
	logger.Info("start cache dump", zap.String("filename", dumpFilename))
	logger.Info("start wal write", zap.String("filename", xlogFilename))

	// open dump file
	dump, err := os.Create(dumpFilename)
	if err != nil {
		return err
	}
	dumpWriter := bufio.NewWriterSize(dump, 1048576) // 1Mb

	// start input dumper
	xlog, err := os.Create(xlogFilename)
	if err != nil {
		return err
	}
	xlogWriter := &SyncWriter{w: bufio.NewWriterSize(xlog, 4096)} // 4kb

	app.Cache.DivertToXlog(xlogWriter)

	// stop cache
	dumpStart := time.Now()
	cacheSize := app.Cache.Size()

	// dump cache
	err = app.Cache.DumpBinary(dumpWriter)
	if err != nil {
		logger.Info("dump failed", zap.Error(err))
		return err
	}

	logger.Info("cache dump finished",
		zap.Int("records", int(cacheSize)),
		zap.Duration("runtime", time.Since(dumpStart)),
	)

	if err = dumpWriter.Flush(); err != nil {
		logger.Info("dump flush failed", zap.Error(err))
		return err
	}

	if err = dump.Close(); err != nil {
		logger.Info("dump close failed", zap.Error(err))
		return err
	}

	// cache dump finished

	logger.Info("dump finished")

	logger.Info("stop listeners")

	stopped := make(chan struct{})

	go func() {
		app.stopListeners()

		if err := xlogWriter.Flush(); err != nil {
			logger.Info("xlog flush failed", zap.Error(err))
			return
		}

		if err := xlog.Close(); err != nil {
			logger.Info("xlog close failed", zap.Error(err))
			return
		}

		close(stopped)
	}()

	select {
	case <-time.After(5 * time.Second):
		logger.Info("stop listeners timeout, force stop daemon")
	case <-stopped:
		logger.Info("listeners stopped")
	}

	// logger.Info("stop all")
	// app.stopAll()

	return nil
}
func ( app * App ) DumpStop ( ) error { app . Lock ( ) defer app . Unlock ( ) if ! app . Config . Dump . Enabled { return nil } if app . Persister != nil { app . Persister . Stop ( ) app . Persister = nil } logger := zapwriter . Logger ( " " ) logger . Info ( " " ) filenamePostfix := fmt . Sprintf ( " " , os . Getpid ( ) , time . Now ( ) . UnixNano ( ) ) dumpFilename := path . Join ( app . Config . Dump . Path , fmt . Sprintf ( " " , filenamePostfix ) ) xlogFilename := path . Join ( app . Config . Dump . Path , fmt . Sprintf ( " " , filenamePostfix ) ) // start dumpers logger . Info ( " " , zap . String ( " " , dumpFilename ) ) logger . Info ( " " , zap . String ( " " , xlogFilename ) ) // open dump file dump , err := os . Create ( dumpFilename ) if err != nil { return err } dumpWriter := bufio . NewWriterSize ( dump , 1048576 ) // 1Mb // start input dumper xlog , err := os . Create ( xlogFilename ) if err != nil { return err } xlogWriter := & SyncWriter { w : bufio . NewWriterSize ( xlog , 4096 ) } // 4kb app . Cache . DivertToXlog ( xlogWriter ) // stop cache dumpStart := time . Now ( ) cacheSize := app . Cache . Size ( ) // dump cache err = app . Cache . DumpBinary ( dumpWriter ) if err != nil { logger . Info ( " " , zap . Error ( err ) ) return err } logger . Info ( " " , zap . Int ( " " , int ( cacheSize ) ) , zap . Duration ( " " , time . Since ( dumpStart ) ) , ) if err = dumpWriter . Flush ( ) ; err != nil { logger . Info ( " " , zap . Error ( err ) ) return err } if err = dump . Close ( ) ; err != nil { logger . Info ( " " , zap . Error ( err ) ) return err } // cache dump finished logger . Info ( " " ) logger . Info ( " " ) stopped := make ( chan struct { } ) go func ( ) { app . stopListeners ( ) if err := xlogWriter . Flush ( ) ; err != nil { logger . Info ( " " , zap . Error ( err ) ) return } if err := xlog . Close ( ) ; err != nil { logger . Info ( " " , zap . Error ( err ) ) return } close ( stopped ) } ( ) select { case <- time . After ( 5 * time . Second ) : logger . Info ( " " ) case <- stopped : logger . Info ( " " ) } // logger.Info("stop all") // app.stopAll() return nil }
// DumpStop implements gracefully stop:
// * Start writing all new data to xlogs
// * Stop cache worker
// * Dump all cache to file
// * Stop listeners
// * Close xlogs
// * Exit application
DumpStop implements gracefully stop : * Start writing all new data to xlogs * Stop cache worker * Dump all cache to file * Stop listeners * Close xlogs * Exit application