}
	if m.subroute != nil {
		msg.Flags |= FlagSubroute
	}
	ch := make(chan Response, 1)
	m.respMu.Lock()
	if m.closed {
		m.respMu.Unlock()
		return nil, ErrDisconnected
	}
	m.respWait = ch
	m.respMu.Unlock()
	ctx := m.ctx

	// Add deadline if none.
	if msg.DeadlineMS == 0 {
		msg.DeadlineMS = uint32(defaultSingleRequestTimeout / time.Millisecond)

	}
}

func (pm *basicManager) GetPods() []*v1.Pod {
	pm.lock.RLock()
	defer pm.lock.RUnlock()
	return podsMapToPods(pm.podByUID)
}

func (pm *basicManager) GetPodsAndMirrorPods() (allPods []*v1.Pod, allMirrorPods []*v1.Pod, orphanedMirrorPodFullnames []string) {
	pm.lock.RLock()
	defer pm.lock.RUnlock()
	allPods = podsMapToPods(pm.podByUID)
	allMirrorPods = mirrorPodsMapToMirrorPods(pm.mirrorPodByUID)

	l.mu.Lock()
	defer l.mu.Unlock()
	cv, ok := l.store.Get(key)
	if !ok || cv.value == nil {
		miss()
		return nil
	}
	if cv.token >= token {
		hit()
		return cv.value
	}
	miss()
	return nil
}

func (l *lruCache[K]) Clear(configs sets.Set[ConfigKey]) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.token = CacheToken(time.Now().UnixNano())

        true        | false       | RETRY
        false       | true        | RETRY
        false       | false       | FINISHED
    }

    def "can unlock resources with unlock"() {
        def lock1 = resourceLock("lock1", lock1Locked, true)
        def lock2 = resourceLock("lock1", lock2Locked, true)

        when:
        def disposition = null

	c.lock.Lock()
	defer c.lock.Unlock()
	err := c.Store.Replace(list, resourceVersion)
	if err != nil {
		return err
	}
	c.initialized = true
	return nil
}

func (c *cacheStore) hasSynced() bool {
	c.lock.Lock()
	defer c.lock.Unlock()
	return c.initialized
}

func (c *cacheStore) unsetInitialized() {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.initialized = false
}

	}

	sys.Lock()
	sys.initialized = true
	sys.Unlock()

	if globalIsDistErasure {
		go sys.refreshBucketsMetadataLoop(ctx, failedBuckets)
	}
}

// Reset the state of the BucketMetadataSys.
func (sys *BucketMetadataSys) Reset() {
	sys.Lock()
	for k := range sys.metadataMap {
		delete(sys.metadataMap, k)
	}
	sys.Unlock()
}

// NewBucketMetadataSys - creates new policy system.

	timestamp time.Time
}

func (a *atomicLastError) Store(err error, t time.Time) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.timestamp.IsZero() || a.timestamp.Before(t) {
		a.err = err
		a.timestamp = t
	}
}

func (a *atomicLastError) Load() error {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.err
}

}

func (plugin *TestPlugin) GetErrorEncountered() bool {
	plugin.pluginLock.RLock()
	defer plugin.pluginLock.RUnlock()
	return plugin.ErrorEncountered
}

func (plugin *TestPlugin) GetAttachedVolumes() map[string][]string {
	plugin.pluginLock.RLock()
	defer plugin.pluginLock.RUnlock()
	ret := make(map[string][]string)
	for nodeName, volumeList := range plugin.attachedVolumeMap {

	f.Lock()
	defer f.Unlock()
	return f.assertList(pods, f.StartedPods)
}

func (f *FakeRuntime) AssertKilledPods(pods []string) bool {
	f.Lock()
	defer f.Unlock()
	return f.assertList(pods, f.KilledPods)
}

func (f *FakeRuntime) AssertStartedContainers(containers []string) bool {
	f.Lock()
	defer f.Unlock()
	return f.assertList(containers, f.StartedContainers)
}

	c.rwMutex.RLock()
	defer c.rwMutex.RUnlock()

	objInfo, err := c.getObjInfo(key)
	if err != nil {
		return nil, err
	}
	return objInfo.latestObj, nil
}

// GetAPIObj gets the informer cache's version by its key.
func (c *AssumeCache) GetAPIObj(key string) (interface{}, error) {
	c.rwMutex.RLock()
	defer c.rwMutex.RUnlock()

	objInfo, err := c.getObjInfo(key)
	if err != nil {