func (bw *baseWorker) Stop() {
	if !bw.isWorkerStarted {
		return
	}
	close(bw.shutdownCh)
	bw.limiterContextCancel()
	if success := util.AwaitWaitGroup(&bw.shutdownWG, bw.options.shutdownTimeout); !success {
		traceLog(func() {
			bw.logger.Info("Worker graceful shutdown timed out.", zap.Duration("Shutdown timeout", bw.options.shutdownTimeout))
		})
	}
	if bw.options.userContextCancel != nil {
		bw.options.userContextCancel()
	}
	return
}
Shutdown is a blocking call and cleans up all the resources associated with worker.
package internal
import (
	"context"
	"errors"
	"fmt"
	"os"
	"sync"
	"syscall"
	"time"
	"github.com/uber-go/tally"
	"go.uber.org/cadence/.gen/go/shared"
	"go.uber.org/cadence/internal/common/backoff"
	"go.uber.org/cadence/internal/common/metrics"
	"go.uber.org/cadence/internal/common/util"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"golang.org/x/time/rate"
)
const (
	retryPollOperationInitialInterval = 20 * time.Millisecond
	retryPollOperationMaxInterval     = 10 * time.Second
)
var (
	pollOperationRetryPolicy = createPollRetryPolicy()
)
var errShutdown = errors.New("worker shutting down")
type (
	resultHandler   func(result []byte, err error)
	laResultHandler func(lar *localActivityResultWrapper)
	localActivityResultWrapper struct {
		err     error
		result  []byte
		attempt int32
		backoff time.Duration
	}
	workflowEnvironment interface {
		asyncActivityClient
		localActivityClient
		workflowTimerClient
		SideEffect(f func() ([]byte, error), callback resultHandler)
		GetVersion(changeID string, minSupported, maxSupported Version) Version
		WorkflowInfo() *WorkflowInfo
		Complete(result []byte, err error)
		RegisterCancelHandler(handler func())
		RequestCancelChildWorkflow(domainName, workflowID string)
		RequestCancelExternalWorkflow(domainName, workflowID, runID string, callback resultHandler)
		ExecuteChildWorkflow(params executeWorkflowParams, callback resultHandler, startedHandler func(r WorkflowExecution, e error)) error
		GetLogger() *zap.Logger
		GetMetricsScope() tally.Scope
		RegisterSignalHandler(handler func(name string, input []byte))
		SignalExternalWorkflow(domainName, workflowID, runID, signalName string, input []byte, arg interface{}, childWorkflowOnly bool, callback resultHandler)
		RegisterQueryHandler(handler func(queryType string, queryArgs []byte) ([]byte, error))
		IsReplaying() bool
		MutableSideEffect(id string, f func() interface{}, equals func(a, b interface{}) bool) Value
		GetDataConverter() DataConverter
		AddSession(sessionInfo *SessionInfo)
		RemoveSession(sessionID string)
		GetContextPropagators() []ContextPropagator
		UpsertSearchAttributes(attributes map[string]interface{}) error
		GetRegistry() *registry
		GetWorkflowInterceptors() []WorkflowInterceptorFactory
	}
	workflowDefinition interface {
		Execute(env workflowEnvironment, header *shared.Header, input []byte)
		OnDecisionTaskStarted()
		StackTrace() string  
		Close()
	}
	baseWorkerOptions struct {
		pollerCount       int
		pollerRate        int
		maxConcurrentTask int
		maxTaskPerSecond  float64
		taskWorker        taskPoller
		identity          string
		workerType        string
		shutdownTimeout   time.Duration
		userContextCancel context.CancelFunc
	}
	baseWorker struct {
		options              baseWorkerOptions
		isWorkerStarted      bool
		shutdownCh           chan struct{}   
		shutdownWG           sync.WaitGroup  
		pollLimiter          *rate.Limiter
		taskLimiter          *rate.Limiter
		limiterContext       context.Context
		limiterContextCancel func()
		retrier              *backoff.ConcurrentRetrier  
		logger               *zap.Logger
		metricsScope         tally.Scope
		pollerRequestCh    chan struct{}
		taskQueueCh        chan interface{}
		sessionTokenBucket *sessionTokenBucket
	}
	polledTask struct {
		task interface{}
	}
)
func createPollRetryPolicy() backoff.RetryPolicy {
	policy := backoff.NewExponentialRetryPolicy(retryPollOperationInitialInterval)
	policy.SetMaximumInterval(retryPollOperationMaxInterval)
	policy.SetExpirationInterval(backoff.NoInterval)  
	return policy
}
func newBaseWorker(options baseWorkerOptions, logger *zap.Logger, metricsScope tally.Scope, sessionTokenBucket *sessionTokenBucket) *baseWorker {
	ctx, cancel := context.WithCancel(context.Background())
	bw := &baseWorker{
		options:         options,
		shutdownCh:      make(chan struct{}),
		taskLimiter:     rate.NewLimiter(rate.Limit(options.maxTaskPerSecond), 1),
		retrier:         backoff.NewConcurrentRetrier(pollOperationRetryPolicy),
		logger:          logger.With(zapcore.Field{Key: tagWorkerType, Type: zapcore.StringType, String: options.workerType}),
		metricsScope:    tagScope(metricsScope, tagWorkerType, options.workerType),
		pollerRequestCh: make(chan struct{}, options.maxConcurrentTask),
		taskQueueCh:     make(chan interface{}),  
		limiterContext:       ctx,
		limiterContextCancel: cancel,
		sessionTokenBucket:   sessionTokenBucket,
	}
	if options.pollerRate > 0 {
		bw.pollLimiter = rate.NewLimiter(rate.Limit(options.pollerRate), 1)
	}
	return bw
}
func (bw *baseWorker) Start() {
	if bw.isWorkerStarted {
		return
	}
	bw.metricsScope.Counter(metrics.WorkerStartCounter).Inc(1)
	for i := 0; i < bw.options.pollerCount; i++ {
		bw.shutdownWG.Add(1)
		go bw.runPoller()
	}
	bw.shutdownWG.Add(1)
	go bw.runTaskDispatcher()
	bw.isWorkerStarted = true
	traceLog(func() {
		bw.logger.Info("Started Worker",
			zap.Int("PollerCount", bw.options.pollerCount),
			zap.Int("MaxConcurrentTask", bw.options.maxConcurrentTask),
			zap.Float64("MaxTaskPerSecond", bw.options.maxTaskPerSecond),
		)
	})
}
func (bw *baseWorker) isShutdown() bool {
	select {
	case <-bw.shutdownCh:
		return true
	default:
		return false
	}
}
func (bw *baseWorker) runPoller() {
	defer bw.shutdownWG.Done()
	bw.metricsScope.Counter(metrics.PollerStartCounter).Inc(1)
	for {
		select {
		case <-bw.shutdownCh:
			return
		case <-bw.pollerRequestCh:
			if bw.sessionTokenBucket != nil {
				bw.sessionTokenBucket.waitForAvailableToken()
			}
			bw.pollTask()
		}
	}
}
func (bw *baseWorker) runTaskDispatcher() {
	defer bw.shutdownWG.Done()
	for i := 0; i < bw.options.maxConcurrentTask; i++ {
		bw.pollerRequestCh <- struct{}{}
	}
	for {
		select {
		case <-bw.shutdownCh:
			return
		case task := <-bw.taskQueueCh:
			_, isPolledTask := task.(*polledTask)
			if isPolledTask && bw.taskLimiter.Wait(bw.limiterContext) != nil {
				if bw.isShutdown() {
					return
				}
			}
			bw.shutdownWG.Add(1)
			go bw.processTask(task)
		}
	}
}
func (bw *baseWorker) pollTask() {
	var err error
	var task interface{}
	bw.retrier.Throttle()
	if bw.pollLimiter == nil || bw.pollLimiter.Wait(bw.limiterContext) == nil {
		task, err = bw.options.taskWorker.PollTask()
		if err != nil && enableVerboseLogging {
			bw.logger.Debug("Failed to poll for task.", zap.Error(err))
		}
		if err != nil {
			if isNonRetriableError(err) {
				bw.logger.Error("Worker received non-retriable error. Shutting down.", zap.Error(err))
				p, _ := os.FindProcess(os.Getpid())
				p.Signal(syscall.SIGINT)
				return
			}
			bw.retrier.Failed()
		} else {
			bw.retrier.Succeeded()
		}
	}
	if task != nil {
		select {
		case bw.taskQueueCh <- &polledTask{task}:
		case <-bw.shutdownCh:
		}
	} else {
		bw.pollerRequestCh <- struct{}{}  
	}
}
func isNonRetriableError(err error) bool {
	if err == nil {
		return false
	}
	switch err.(type) {
	case *shared.BadRequestError,
		*shared.ClientVersionNotSupportedError:
		return true
	}
	return false
}
func (bw *baseWorker) processTask(task interface{}) {
	defer bw.shutdownWG.Done()
	polledTask, isPolledTask := task.(*polledTask)
	if isPolledTask {
		task = polledTask.task
	}
	defer func() {
		if p := recover(); p != nil {
			bw.metricsScope.Counter(metrics.WorkerPanicCounter).Inc(1)
			topLine := fmt.Sprintf("base worker for %s [panic]:", bw.options.workerType)
			st := getStackTraceRaw(topLine, 7, 0)
			bw.logger.Error("Unhandled panic.",
				zap.String(tagPanicError, fmt.Sprintf("%v", p)),
				zap.String(tagPanicStack, st))
		}
		if isPolledTask {
			bw.pollerRequestCh <- struct{}{}
		}
	}()
	err := bw.options.taskWorker.ProcessTask(task)
	if err != nil {
		if isClientSideError(err) {
			bw.logger.Info("Task processing failed with client side error", zap.Error(err))
		} else {
			bw.logger.Info("Task processing failed with error", zap.Error(err))
		}
	}
}
func (bw *baseWorker) Run() {
	bw.Start()
	d := <-getKillSignal()
	traceLog(func() {
		bw.logger.Info("Worker has been killed", zap.String("Signal", d.String()))
	})
	bw.Stop()
}