func (ahs *allHealState) setDiskHealingStatus(ep Endpoint, healing bool) {
	ahs.Lock()
	defer ahs.Unlock()

	ahs.healLocalDisks[ep] = healing
}

func (ahs *allHealState) pushHealLocalDisks(healLocalDisks ...Endpoint) {
	ahs.Lock()
	defer ahs.Unlock()

	for _, ep := range healLocalDisks {
		ahs.healLocalDisks[ep] = false
	}
}

func (ahs *allHealState) periodicHealSeqsClean(ctx context.Context) {

	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()
	delete(r.Cache, bucket)
}

// UpdateReplicaStat updates in-memory replica statistics with new values.
func (r *ReplicationStats) UpdateReplicaStat(bucket string, n int64) {
	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()
	bs, ok := r.Cache[bucket]
	if !ok {
		bs = newBucketReplicationStats()

		}
		cancel(ErrDisconnected)
		c.connChange.L.Lock()
		if atomic.CompareAndSwapUint32((*uint32)(&c.state), StateConnected, StateConnectionError) {
			c.connChange.Broadcast()
		}
		c.disconnected()
		c.connChange.L.Unlock()

		conn.Close()
		c.handleMsgWg.Done()
	}()

	c.connMu.Lock()
	connPingInterval := c.connPingInterval
	c.connMu.Unlock()
	ping := time.NewTicker(connPingInterval)

func (p *podNetnsCache) Get(uid string) Netns {
	// lock current snapshot pod map
	p.mu.RLock()
	defer p.mu.RUnlock()
	if info, f := p.currentPodCache[uid]; f {
		return info.Netns
	}
	return nil
}

// make sure uid is in the cache, even if we don't have a netns
func (p *podNetnsCache) Ensure(uid string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if _, ok := p.currentPodCache[uid]; !ok {

		if t.updating.TryLock() {
			go func() {
				defer t.updating.Unlock()
				t.update()
			}()
		}
		return v.v, v.e
	}

	// Get lock. Either we get it or we wait for it.
	t.updating.Lock()
	if time.Since(time.UnixMilli(t.lastUpdateMs.Load())) < ttl {
		// There is a new value, release lock and return it.
		v = t.valErr.Load()
		t.updating.Unlock()
		return v.v, v.e
	}
	t.update()
	v = t.valErr.Load()

	sys.aMutex.Lock()
	defer sys.aMutex.Unlock()
	if v, ok := sys.arnErrsMap[arn]; !ok {
		sys.arnErrsMap[arn] = arnErrs{
			updateInProgress: true,
			count:            v.count + 1,
			bucket:           bucket,
		}
	}
}

func (sys *BucketTargetSys) markRefreshDone(bucket, arn string) {
	sys.aMutex.Lock()
	defer sys.aMutex.Unlock()
	if v, ok := sys.arnErrsMap[arn]; ok {

	list.statLock.Lock()
	defer list.statLock.Unlock()

	stats, ok := list.targetStats[id]
	if !ok {
		// should not happen
		return
	}

	stats.currentSendCalls--
	list.targetStats[id] = stats
	return
}

func (list *TargetList) incFailedEvents(id TargetID) {
	list.statLock.Lock()
	defer list.statLock.Unlock()

	stats, ok := list.targetStats[id]
	if !ok {

			ok = true
		}
	}
	if ok {
		lock := z.serverPools[0].NewNSLock(minioMetaBucket, rebalMetaName)
		lkCtx, err := lock.GetLock(ctx, globalOperationTimeout)
		if err != nil {
			rebalanceLogIf(ctx, fmt.Errorf("failed to acquire write lock on %s/%s: %w", minioMetaBucket, rebalMetaName, err))
			return err
		}
		defer lock.Unlock(lkCtx)

		ctx = lkCtx.Context()

	if responses == nil {
		return nil, errors.New("RequestStream: responses channel is nil")
	}
	m.init = true
	m.respMu.Lock()
	if m.closed {
		m.respMu.Unlock()
		return nil, ErrDisconnected
	}
	m.respWait = responses // Route directly to output.
	m.respMu.Unlock()

	// Try to grab an initial block.
	m.singleResp = false
	m.RecvSeq = m.SendSeq // Sync
	if cap(requests) > 0 {

type IAMStorageAPI interface {
	// The role of the read-write lock is to prevent go routines from
	// concurrently reading and writing the IAM storage. The (r)lock()
	// functions return the iamCache. The cache can be safely written to
	// only when returned by `lock()`.
	lock() *iamCache
	unlock()
	rlock() *iamCache
	runlock()
	getUsersSysType() UsersSysType