func getDrainTimeout(ctx thrift.Context) time.Duration {
	if ctx != nil {
		if deadline, ok := ctx.Deadline(); ok {
			return deadline.Sub(time.Now())
		}
	}
	return defaultDrainTimeout
}
determine how long we need to wait for the wait group
if the thrift context timeout is set and is smaller than the default timeout, we
should use that.
package inputhost
import (
	"fmt"
	"net/http"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/tchannel-go/thrift"
	ccommon "github.com/uber/cherami-client-go/common"
	"github.com/uber/cherami-server/common"
	dconfig "github.com/uber/cherami-server/common/dconfigclient"
	mm "github.com/uber/cherami-server/common/metadata"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-server/services/inputhost/load"
	"github.com/uber/cherami-server/stream"
	"github.com/uber/cherami-thrift/.generated/go/admin"
	"github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
)
const (
	defaultIdleTimeout = 10 * time.Minute
	defaultBufferSize = 1000
	hostLoadReportingInterval = 2 * time.Second
	minimumAllowedMessageDelaySeconds = 60
	minimumAllowedMessageDelaySecondsTest = 3
	extentRolloverSeqnumMin, extentRolloverSeqnumMax = 10000000, 20000000
	defaultDrainTimeout = 1 * time.Minute
	connWGTimeout = 5 * time.Second
	drainAllUUID = "D3A9C5DC-AE62-4465-9898-7FE71BD1FCA"
)
var (
	batchMsgAckTimeout = 1 * time.Minute  
)
type (
	InputHost struct {
		logger                 bark.Logger
		loadShutdownRef        int32
		pathMutex              sync.RWMutex
		pathCache              map[string]*inPathCache
		pathCacheByDestPath    map[string]string
		shutdownWG             sync.WaitGroup
		shutdown               chan struct{}
		cacheTimeout           time.Duration
		mClient                metadata.TChanMetadataService
		hostIDHeartbeater      common.HostIDHeartbeater
		loadReporter           common.LoadReporterDaemon
		m3Client               metrics.Client
		dConfigClient          dconfig.Client
		tokenBucketValue       atomic.Value  
		hostConnLimit          int32
		hostConnLimitPerSecond int32
		maxConnLimit           int32
		extMsgsLimitPerSecond  int32
		connMsgsLimitPerSecond int32
		hostMetrics            *load.HostMetrics
		lastLoadReportedTime   int64         
		nodeStatus             atomic.Value  
		testShortExtentsByPath string        
		common.SCommon
	}
	InOptions struct {
		CacheIdleTimeout time.Duration
	}
	extentInfo struct {
		uuid     string
		replicas []string
	}
	inReconfigInfo struct {
		req        *admin.DestinationUpdatedNotification
		updateUUID string
	}
)
var _ cherami.TChanBIn = &InputHost{}
var _ admin.TChanInputHostAdmin = &InputHost{}
var _ common.WSService = &InputHost{}
var ErrHostShutdown = &cherami.InternalServiceError{Message: "InputHost already shutdown"}
var ErrThrottled = &cherami.InternalServiceError{Message: "InputHost throttling publisher cconnection"}
var ErrDstNotLoaded = &cherami.InternalServiceError{Message: "Destination no longer served by this input host"}
var ErrDrainTimedout = &cherami.InternalServiceError{Message: "Draining of Extents timedout"}
func (h *InputHost) isDestinationWritable(destDesc *shared.DestinationDescription) bool {
	status := destDesc.GetStatus()
	if status != shared.DestinationStatus_ENABLED && status != shared.DestinationStatus_SENDONLY {
		return false
	}
	return true
}
func (h *InputHost) listExtentsByStatus(ctx thrift.Context, destUUID string, status *shared.ExtentStatus) (*metadata.ListInputHostExtentsStatsResult_, error) {
	listReq := metadata.ListInputHostExtentsStatsRequest{
		InputHostUUID:   common.StringPtr(h.GetHostUUID()),
		DestinationUUID: common.StringPtr(destUUID),
		Status:          status,
	}
	listResp, err := h.mClient.ListInputHostExtentsStats(ctx, &listReq)
	if err != nil {
		return nil, fmt.Errorf("ListInputHostExtentStats() returned error dst=%v, err=%v",
			destUUID, err)
	}
	if listResp == nil || len(listResp.GetExtentStatsList()) == 0 {
		return nil, fmt.Errorf("getExtentsInfoForDestination - Found zero extents for destination, dst=%v",
			destUUID)
	}
	return listResp, nil
}
func (h *InputHost) createExtentInfo(extentUUID string, storeIDs []string) *extentInfo {
	if extentUUID == "" || storeIDs == nil || len(storeIDs) == 0 {
		return nil
	}
	var rpm = h.GetRingpopMonitor()
	var info = &extentInfo{
		uuid:     extentUUID,
		replicas: make([]string, 0, 3),
	}
	for _, storeID := range storeIDs {
		addr, err := rpm.ResolveUUID(common.StoreServiceName, storeID)
		if err != nil {
			h.logger.WithField(`storeID`, storeID).
				WithField(common.TagErr, err).
				Error("Cannot resolve UUID for store")
			return nil
		}
		info.replicas = append(info.replicas, addr)
	}
	return info
}
func (h *InputHost) getExtentsInfoForDestination(ctx thrift.Context, destUUID string) ([]*extentInfo, error) {
	listResp, err := h.listExtentsByStatus(ctx, destUUID, common.MetadataExtentStatusPtr(shared.ExtentStatus_OPEN))
	if err != nil {
		return nil, err
	}
	var extents = make([]*extentInfo, 0, 2)
	for _, item := range listResp.ExtentStatsList {
		extent := item.Extent
		info := h.createExtentInfo(extent.GetExtentUUID(), extent.StoreUUIDs)
		if info == nil {
			continue
		}
		extents = append(extents, info)
	}
	if len(extents) < 1 {
		h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
			WithField("nExtents", len(listResp.GetExtentStatsList())).
			Error("Can't find extents for destination, no healthy extent found")
		return nil, fmt.Errorf("No healthy extent found")
	}
	return extents, nil
}
func (h *InputHost) getExtentsAndLoadPathCache(ctx thrift.Context, destPath, destUUID string, destType shared.DestinationType) (*inPathCache, error) {
	extents, err := h.getExtentsInfoForDestination(ctx, destUUID)
	if err != nil {
		return nil, err
	}
	pathCache := h.loadPath(extents, destPath, destUUID, destType, h.m3Client)
	return pathCache, nil
}
func (h *InputHost) checkDestination(ctx thrift.Context, path string) (string, shared.DestinationType, metrics.ErrorClass, error) {
	mGetRequest := shared.ReadDestinationRequest{Path: common.StringPtr(path)}
	destDesc, err := h.mClient.ReadDestination(ctx, &mGetRequest)
	if err != nil {
		errC, newErr := common.ConvertDownstreamErrors(h.logger, err)
		return "", shared.DestinationType_UNKNOWN, errC, newErr
	}
	if destDesc == nil {
		errMsg := fmt.Sprintf("unable to get destination description from metadata for dst=%v", path)
		errC, newErr := common.ConvertDownstreamErrors(h.logger, &cherami.BadRequestError{Message: errMsg})
		return "", shared.DestinationType_UNKNOWN, errC, newErr
	}
	if !h.isDestinationWritable(destDesc) {
		errMsg := fmt.Sprintf("Destination is not writable, dst=%v, status=%v", path, destDesc.GetStatus())
		errC, newErr := common.ConvertDownstreamErrors(h.logger, &cherami.BadRequestError{Message: errMsg})
		return "", shared.DestinationType_UNKNOWN, errC, newErr
	}
	return destDesc.GetDestinationUUID(), destDesc.GetType(), metrics.NoError, nil
}
func (h *InputHost) OpenPublisherStreamHandler(w http.ResponseWriter, r *http.Request) {
	path := r.Header.Get("path")
	if len(path) == 0 {
		err := &cherami.BadRequestError{}
		h.logger.Warn("please set the path as part of the header. BadRequestError")
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	wsStream, err := h.GetWSConnector().AcceptPublisherStream(w, r)
	if err != nil {
		h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
			WithField(common.TagErr, err).Error("unable to upgrade websocket connection")
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostInternalFailures)
		return
	}
	ctx, cancel := thrift.NewContext(common.MaxDuration)
	defer cancel()
	ctx = thrift.WithHeaders(ctx, map[string]string{
		"path": path,
	})
	if err = h.OpenPublisherStream(ctx, wsStream); err != nil {
		h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
			WithField(common.TagErr, err).Error("unable to open publish stream")
		return
	}
}
func (h *InputHost) OpenPublisherStream(ctx thrift.Context, call stream.BInOpenPublisherStreamInCall) error {
	h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostRequests)
	path, ok := (ctx.Headers()["path"])
	if !ok {
		err := &cherami.BadRequestError{}
		h.logger.Warn("please set the path as part of the header. BadRequestError")
		call.Done()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		return err
	}
	h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).Debug("inputhost: OpenPublisherStream called with path")
	if h.IsLimitsEnabled() {
		if ok, _ = h.GetTokenBucketValue().TryConsume(1); !ok {
			h.logger.WithField(common.TagHostConnLimit,
				common.FmtHostConnLimit(h.GetHostConnLimitPerSecond())).
				Warn("Too many open publisher streams on this inputhost within a second")
			call.Done()
			h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
			return ErrThrottled
		}
	}
	if h.IsLimitsEnabled() && (h.GetNumConnections() > h.GetHostConnLimitOverall()) {
		h.logger.WithField(common.TagHostConnLimit,
			common.FmtHostConnLimit(h.GetNumConnections())).
			Warn("Too many open connections on this inputhost. Rejecting this open")
		call.Done()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		return ErrThrottled
	}
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		atomic.AddInt32(&h.loadShutdownRef, -1)
		_, newErr := common.ConvertDownstreamErrors(h.logger, ErrHostShutdown)
		call.Done()
		return newErr
	}
	pathCache, ok := h.getPathCacheByDestPath(path)
	if !ok || pathCache == nil {
		destinationUUID, destType, errC, err := h.checkDestination(ctx, path)
		if err != nil {
			h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
				WithField(common.TagErr, err).Error("failed on destination check")
			atomic.AddInt32(&h.loadShutdownRef, -1)
			call.Done()
			h.incFailureCounter(metrics.OpenPublisherStreamScope, errC)
			return err
		}
		pathCache, err = h.getExtentsAndLoadPathCache(ctx, path, destinationUUID, destType)
		if err != nil || pathCache == nil {
			h.logger.WithField(common.TagDstPth, common.FmtDstPth(path)).
				WithField(common.TagErr, err).Error("unable to load path cache for destination")
			atomic.AddInt32(&h.loadShutdownRef, -1)
			call.Done()
			h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostInternalFailures)
			return &ReplicaNotExistsError{}
		}
	}
	doneCh := make(chan bool, 5)
	pathCache.Lock()
	errCleanup := func() {
		pathCache.Unlock()
		atomic.AddInt32(&h.loadShutdownRef, -1)
		call.Done()
	}
	if !pathCache.isActive() {
		errCleanup()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostInternalFailures)
		return ErrDstNotLoaded
	}
	hostMaxConnPerDestination := h.GetMaxConnPerDest()
	if h.IsLimitsEnabled() && pathCache.dstMetrics.Get(load.DstMetricNumOpenConns) > int64(hostMaxConnPerDestination) {
		pathCache.logger.WithField(common.TagHostConnLimit,
			common.FmtHostConnLimit(hostMaxConnPerDestination)).
			Warn("Too many open connections on this path. Rejecting this open")
		errCleanup()
		h.m3Client.IncCounter(metrics.OpenPublisherStreamScope, metrics.InputhostUserFailures)
		return ErrThrottled
	}
	conn := newPubConnection(path, call, pathCache, h.m3Client, h.IsLimitsEnabled(), h.cacheTimeout, doneCh)
	pathCache.connections[pathCache.currID] = conn
	conn.open()
	pathCache.currID++
	pathCache.dstMetrics.Increment(load.DstMetricNumOpenConns)
	pathCache.Unlock()
	h.hostMetrics.Increment(load.HostMetricNumOpenConns)
	atomic.AddInt32(&h.loadShutdownRef, -1)
	<-conn.doneCh
	pathCache.dstMetrics.Decrement(load.DstMetricNumOpenConns)
	h.hostMetrics.Decrement(load.HostMetricNumOpenConns)
	return nil
}
func (h *InputHost) PutMessageBatch(ctx thrift.Context, request *cherami.PutMessageBatchRequest) (*cherami.PutMessageBatchResult_, error) {
	sw := h.m3Client.StartTimer(metrics.PutMessageBatchInputHostScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostRequests)
	path := request.GetDestinationPath()
	messages := request.GetMessages()
	lclLg := h.logger.WithField(common.TagDstPth, common.FmtDstPth(path))
	var msgLength int64
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		atomic.AddInt32(&h.loadShutdownRef, -1)
		_, newErr := common.ConvertDownstreamErrors(h.logger, ErrHostShutdown)
		return nil, newErr
	}
	pathCache, ok := h.getPathCacheByDestPath(path)
	if !ok || pathCache == nil {
		destinationUUID, destType, errC, err := h.checkDestination(ctx, path)
		if err != nil {
			lclLg.WithField(common.TagErr, err).Error("failed on destination check")
			h.incFailureCounter(metrics.PutMessageBatchInputHostScope, errC)
			return nil, err
		}
		pathCache, err = h.getExtentsAndLoadPathCache(ctx, path, destinationUUID, destType)
		if err != nil || pathCache == nil {
			lclLg.WithField(common.TagErr, err).Error("unable to load path cache for destination")
			atomic.AddInt32(&h.loadShutdownRef, -1)
			h.m3Client.IncCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostInternalFailures)
			return nil, &ReplicaNotExistsError{}
		}
	}
	result := cherami.NewPutMessageBatchResult_()
	ackChannel := make(chan *cherami.PutMessageAck, defaultBufferSize)
	inflightRequestCnt := 0
	inflightMsgMap := make(map[string]struct{})
	for _, msg := range messages {
		msgLength = int64(len(msg.Data))
		pathCache.dstMetrics.Add(load.DstMetricBytesIn, msgLength)
		pathCache.hostMetrics.Add(load.HostMetricBytesIn, msgLength)
		inMsg := &inPutMessage{
			putMsg:         msg,
			putMsgAckCh:    ackChannel,
			putMsgRecvTime: time.Now(),
		}
		select {
		case pathCache.putMsgCh <- inMsg:
			inflightRequestCnt++
			inflightMsgMap[msg.GetID()] = struct{}{}
		default:
			result.FailedMessages = append(result.FailedMessages, &cherami.PutMessageAck{
				ID:      common.StringPtr(msg.GetID()),
				Status:  common.CheramiStatusPtr(cherami.Status_THROTTLED),
				Message: common.StringPtr("throttling; inputhost is busy"),
			})
		}
	}
	internalErrs, userErrs := int64(0), int64(0)
	var respStatus cherami.Status
	var respMsg string
	ackReceived := func(ack *cherami.PutMessageAck) {
		if ack.GetStatus() != cherami.Status_OK {
			if ack.GetStatus() != cherami.Status_THROTTLED {
				internalErrs++
			} else {
				userErrs++
			}
			result.FailedMessages = append(result.FailedMessages, ack)
		} else {
			result.SuccessMessages = append(result.SuccessMessages, ack)
		}
		delete(inflightMsgMap, ack.GetID())
	}
	msgTimer := common.NewTimer(batchMsgAckTimeout)
	defer msgTimer.Stop()
ACKDRAIN:
	for i := 0; i < inflightRequestCnt; i++ {
		select {
		case ack := <-ackChannel:
			ackReceived(ack)
		default:
			select {
			case ack := <-ackChannel:
				ackReceived(ack)
			case <-pathCache.closeCh:
				respStatus = cherami.Status_FAILED
				respMsg = "pathCache unloaded"
				break ACKDRAIN
			case <-msgTimer.C:
				respStatus = cherami.Status_TIMEDOUT
				respMsg = "message timedout"
				break ACKDRAIN
			}
		}
	}
	if len(inflightMsgMap) > 0 {
		pathCache.logger.WithFields(bark.Fields{
			`numFailedMessages`: len(inflightMsgMap),
			`respMsg`:           respMsg,
		}).Info("failing putMessageBatch")
		for id := range inflightMsgMap {
			result.FailedMessages = append(result.FailedMessages, &cherami.PutMessageAck{
				ID:      common.StringPtr(id),
				Status:  common.CheramiStatusPtr(respStatus),
				Message: common.StringPtr(respMsg),
			})
			internalErrs++
		}
	}
	pathCache.updateLastDisconnectTime()
	atomic.AddInt32(&h.loadShutdownRef, -1)
	h.m3Client.AddCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostMessageReceived, int64(len(result.SuccessMessages)))
	pathCache.destM3Client.AddCounter(metrics.PutMessageBatchInputHostDestScope, metrics.InputhostDestMessageReceived, int64(len(result.SuccessMessages)))
	h.m3Client.AddCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostMessageUserFailures, userErrs)
	pathCache.destM3Client.AddCounter(metrics.PutMessageBatchInputHostDestScope, metrics.InputhostDestMessageUserFailures, userErrs)
	h.m3Client.AddCounter(metrics.PutMessageBatchInputHostScope, metrics.InputhostMessageInternalFailures, internalErrs)
	pathCache.destM3Client.AddCounter(metrics.PutMessageBatchInputHostDestScope, metrics.InputhostDestMessageInternalFailures, internalErrs)
	pathCache.dstMetrics.Add(load.DstMetricOverallNumMsgs, int64(len(messages)))
	pathCache.dstMetrics.Add(load.DstMetricNumFailed, (internalErrs + userErrs))
	pathCache.dstMetrics.Add(load.DstMetricNumAcks, int64(len(result.SuccessMessages)))
	return result, nil
}
func (h *InputHost) DestinationsUpdated(ctx thrift.Context, request *admin.DestinationsUpdatedRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.DestinationsUpdatedScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.DestinationsUpdatedScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(request.GetUpdateUUID())).Error("inputhost: DestinationsUpdated: dropping reconfiguration due to shutdown")
		h.m3Client.IncCounter(metrics.DestinationsUpdatedScope, metrics.InputhostFailures)
		return ErrHostShutdown
	}
	var intErr error
	updateUUID := request.GetUpdateUUID()
	h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(updateUUID)).
		Debug("inputhost: DestinationsUpdated: processing reconfiguration")
	for _, req := range request.Updates {
		destUUID := req.GetDestinationUUID()
		pathCache, ok := h.getPathCacheByDestUUID(destUUID)
		if ok {
			if pathCache.isActiveNoLock() {
				pathCache.reconfigureCh <- inReconfigInfo{req: req, updateUUID: updateUUID}
			} else {
				intErr = errPathCacheUnloading
			}
		} else {
			intErr = &cherami.EntityNotExistsError{}
		}
		if intErr != nil {
			err = intErr
			h.m3Client.IncCounter(metrics.DestinationsUpdatedScope, metrics.InputhostFailures)
			h.logger.WithFields(bark.Fields{
				common.TagDst:           common.FmtDst(destUUID),
				common.TagReconfigureID: common.FmtReconfigureID(updateUUID),
				common.TagErr:           intErr,
			}).Error("inputhost: DestinationsUpdated: dropping reconfiguration")
		}
	}
	h.logger.WithField(common.TagReconfigureID, common.FmtReconfigureID(updateUUID)).
		Debug("inputhost: DestinationsUpdated: finished reconfiguration")
	return
}
func (h *InputHost) UnloadDestinations(ctx thrift.Context, request *admin.UnloadDestinationsRequest) (err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.UnloadDestinationsScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.UnloadDestinationsScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("not unloading the path cache; inputHost already shutdown")
		h.m3Client.IncCounter(metrics.UnloadDestinationsScope, metrics.InputhostFailures)
		return ErrHostShutdown
	}
	for _, destUUID := range request.DestUUIDs {
		h.pathMutex.RLock()
		pathCache, ok := h.pathCache[destUUID]
		if ok {
			pathCache.Lock()
			pathCache.prepareForUnload()
			go pathCache.unload()
			pathCache.Unlock()
		} else {
			h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
				Error("destination is not cached at all")
			err = ErrDstNotLoaded
			h.m3Client.IncCounter(metrics.UnloadDestinationsScope, metrics.InputhostFailures)
		}
		h.pathMutex.RUnlock()
	}
	return err
}
func (h *InputHost) ListLoadedDestinations(ctx thrift.Context) (result *admin.ListLoadedDestinationsResult_, err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.ListLoadedDestinationsScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.ListLoadedDestinationsScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("inputHost already shutdown")
		h.m3Client.IncCounter(metrics.ListLoadedDestinationsScope, metrics.InputhostFailures)
		return nil, ErrHostShutdown
	}
	result = admin.NewListLoadedDestinationsResult_()
	h.pathMutex.RLock()
	result.Dests = make([]*admin.Destinations, len(h.pathCache))
	count := 0
	for destUUID, pathCache := range h.pathCache {
		destRes := admin.NewDestinations()
		destRes.DestUUID = common.StringPtr(destUUID)
		destRes.DestPath = common.StringPtr(pathCache.destinationPath)
		result.Dests[count] = destRes
		count++
	}
	h.pathMutex.RUnlock()
	return result, err
}
func (h *InputHost) ReadDestState(ctx thrift.Context, request *admin.ReadDestinationStateRequest) (result *admin.ReadDestinationStateResult_, err error) {
	defer atomic.AddInt32(&h.loadShutdownRef, -1)
	sw := h.m3Client.StartTimer(metrics.ReadDestStateScope, metrics.InputhostLatencyTimer)
	defer sw.Stop()
	h.m3Client.IncCounter(metrics.ReadDestStateScope, metrics.InputhostRequests)
	if atomic.AddInt32(&h.loadShutdownRef, 1) <= 0 {
		h.logger.Error("inputHost already shutdown")
		h.m3Client.IncCounter(metrics.ReadDestStateScope, metrics.InputhostFailures)
		return nil, ErrHostShutdown
	}
	result = admin.NewReadDestinationStateResult_()
	result.InputHostUUID = common.StringPtr(h.GetHostUUID())
	result.DestState = make([]*admin.DestinationState, 0)
	h.pathMutex.RLock()
	for _, destUUID := range request.DestUUIDs {
		pathCache, ok := h.pathCache[destUUID]
		if ok {
			destState := pathCache.getState()
			result.DestState = append(result.DestState, destState)
		} else {
			h.logger.WithField(common.TagDst, common.FmtDst(destUUID)).
				Warn("destination is not cached at all")
			err = ErrDstNotLoaded
		}
	}
	h.pathMutex.RUnlock()
	if err != nil {
		h.m3Client.IncCounter(metrics.ReadDestStateScope, metrics.InputhostFailures)
	}
	return result, err
}