func (t *retMgrRunner) stopRetentionMgr() {
	if atomic.CompareAndSwapUint32(&t.running, 1, 0) {
		t.retentionMgr.Stop()
		t.log.Debug("retMgrRunner: Stopped retention manager")
	}
}
stopRetentionMgr: stops retention manager, if it were running
package controllerhost
import (
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/retentionmgr"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
)
const (
	retentionMgrInterval             = 10 * time.Minute
	dlqRetentionMgrInterval          = 1 * time.Hour
	singleCGVisibleExtentGracePeriod = 7 * 24 * time.Hour
	extentDeleteDeferPeriod          = 3 * time.Hour
	retentionMgrWorkers              = 8
)
type (
	retMgrRunnerContext struct {
		hostID         string
		ringpop        common.RingpopMonitor
		metadataClient metadata.TChanMetadataService
		clientFactory  common.ClientFactory
		log            bark.Logger
		m3Client       metrics.Client
		localZone      string
	}
	retMgrRunner struct {
		*retMgrRunnerContext
		retentionMgr   *retentionmgr.RetentionManager
		running        uint32
		listenC        chan *common.RingpopListenerEvent
		stopC          chan struct{}
		sync.WaitGroup  
	}
)
func newRetMgrRunner(ctx *retMgrRunnerContext) *retMgrRunner {
	return &retMgrRunner{
		retMgrRunnerContext: ctx,
		listenC:             make(chan *common.RingpopListenerEvent, 32),
		stopC:               make(chan struct{}),
	}
}
func (t *retMgrRunner) Start() {
	t.log.Debug("retMgrRunner: Starting")
	t.Add(1)
	go t.worker()
}
func (t *retMgrRunner) Stop() {
	close(t.stopC)
	t.Wait()
	t.log.Debug("retMgrRunner: Stopped")
}
func (t *retMgrRunner) worker() {
	defer t.Done()
	err := t.ringpop.AddListener(common.ControllerServiceName, common.ControllerServiceName+"-retention", t.listenC)
	if err != nil {
		t.log.WithField(common.TagErr, err).Error(`retMgrRunner: AddListener('controller', 'controller-retention') failed`)
	}
	if t.isPrimary() {
		t.startRetentionMgr()
	} else {
		t.log.Debug("retMgrRunner: not primary; not running here")
	}
forever:
	for {
		select {
		case <-t.listenC:  
			if t.isPrimary() {
				t.startRetentionMgr()
			} else {
				t.stopRetentionMgr()
			}
		case <-t.stopC:  
			t.log.Debug("retMgrRunner: Stop notification")
			t.stopRetentionMgr()
			break forever
		}
	}
}
func (t *retMgrRunner) isPrimary() bool {
	hostInfo, err := t.ringpop.FindHostForKey(common.ControllerServiceName, "retention-manager")
	if err != nil {
		t.log.WithField(common.TagErr, err).Error(`retMgrRunner: FindHostForKey failed`)
		return true  
	}
	return t.hostID == hostInfo.UUID
}
func (t *retMgrRunner) startRetentionMgr() {
	if atomic.CompareAndSwapUint32(&t.running, 0, 1) {
		if t.retentionMgr == nil {
			opts := &retentionmgr.Options{
				RetentionInterval: retentionMgrInterval,
				SingleCGVisibleExtentGracePeriod: singleCGVisibleExtentGracePeriod,
				ExtentDeleteDeferPeriod:          extentDeleteDeferPeriod,
				NumWorkers:                       retentionMgrWorkers,
				LocalZone:                        t.retMgrRunnerContext.localZone,
			}
			t.retentionMgr = retentionmgr.New(opts, t.metadataClient, t.clientFactory, t.m3Client, t.log)
		}
		t.log.Debug("retMgrRunner: Starting retention manager")
		t.retentionMgr.Start()
	}
}