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
}
// 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