// runLocked runs the main main maintenance loop.
//
// As the run is executed, stats can be collected in ms.st. These will be output
// to datastore on completion.
516529
func (ms *managerShard) runLocked(c context.Context) error {
	workers := ms.cache.Parallel
	if workers <= 0 {
		workers = 1
	}

	// NOTE: This does not currently restrain itself to the current shard. This
	// can be done using a "__key__" inequality filter to bound the query.
	prototype := entry{
		CacheName: ms.cache.Name,
	}
	q := datastore.NewQuery(prototype.kind())

	var (
		totalEntries   = 0
		totalRefreshed = 0
		totalPruned    = 0
		totalErrors    = int32(0)

		entries   = make([]*entry, 0, ms.queryBatchSize)
		putBuf    = make([]*entry, ms.queryBatchSize)
		deleteBuf = make([]*entry, ms.queryBatchSize)
	)

	// Calculate our pruning threshold. If an entry's "LastAccessed" is <= this
	// threshold, it is candidate for pruning.
	var pruneThreshold time.Time
	if pi := ms.cache.pruneInterval(); pi > 0 {
		pruneThreshold = ms.now.Add(-pi)
	}

	// handleEntries refreshes the accumulated entries. It is used as a callback
	// in between query batches, as well as a finalizer after the queries
	// complete.
	//
	// handleEntries is, itself, not goroutine-safe.
	//
	// It will refresh in parallel, adding entries to "putBuf" or "deleteBuf" as
	// appropriate. At the end of its operation, all deferred datastore
	// operations will execute, and the accumulated entries list will be purged
	// for next round.
	handleEntries := func(c context.Context) error {
		if len(entries) == 0 {
			return nil
		}

		// Use atomic-friendly int32 values to index the put/delete buffers. This
		// will let our parallel goroutines safely add entries with very low
		// overhead.
		var putIdx, deleteIdx int32
		putEntry := func(e *entry) { putBuf[atomic.AddInt32(&putIdx, 1)-1] = e }
		deleteEntry := func(e *entry) { deleteBuf[atomic.AddInt32(&deleteIdx, 1)-1] = e }

		// Process each entry in parallel.
		//
		// Each task will return a nil error, so the error result does not need to
		// be observed.
		_ = parallel.WorkPool(workers, func(taskC chan<- func() error) {
			for _, e := range entries {
				e := e

				taskC <- func() error {
					// Is this entry candidate for pruning?
					if !(pruneThreshold.IsZero() || e.LastAccessed.After(pruneThreshold)) {
						log.Fields{
							"key":            e.keyHash(),
							"lastRefresh":    e.LastRefreshed,
							"lastAccessed":   e.LastAccessed,
							"pruneThreshold": pruneThreshold,
						}.Infof(c, "Pruning expired cache entry.")
						deleteEntry(e)
						return nil
					}

					// Is this cache entry candidate for refresh?
					if ms.h != nil {
						refreshInterval := ms.h.RefreshInterval(e.Key)
						if refreshInterval > 0 && !e.LastRefreshed.After(ms.now.Add(-refreshInterval)) {
							// Call our Handler's Refresh function. We leave our cache namespace
							// first.
							switch value, delta, err := doRefresh(c, ms.h, e, ""); err {
							case nil:
								// Refresh successful! Update the entry.
								//
								// Even if he data hasn't changed, the LastRefreshed time has, and
								// the cost of the "Put" is the same either way.
								e.LastRefreshed = ms.now
								e.LastRefreshDelta = int64(delta)
								e.loadValue(value)
								putEntry(e)
								return nil

							case ErrDeleteCacheEntry:
								log.Fields{
									"key": e.keyHash(),
								}.Debugf(c, "Refresh requested entry deletion.")
								deleteEntry(e)
								return nil

							default:
								log.Fields{
									log.ErrorKey:  err,
									"key":         e.keyHash(),
									"lastRefresh": e.LastRefreshed,
								}.Errorf(c, "Failed to refresh cache entry.")
								atomic.AddInt32(&totalErrors, 1)
							}
						}
					}
					return nil
				}
			}
		})

		// Clear our entries buffer for next round.
		entries = entries[:0]

		// Flush our put/delete buffers. Accumulate errors (best effort) and return
		// them as a batch.
		//
		// A failure here is a datastore failure, and will halt processing by
		// propagating from the callback to the query return value.
		_ = parallel.FanOutIn(func(taskC chan<- func() error) {
			if putIdx > 0 {
				taskC <- func() error {
					if err := datastore.Put(c, putBuf[:putIdx]); err != nil {
						log.Fields{
							log.ErrorKey: err,
							"size":       putIdx,
						}.Errorf(c, "Failed to Put batch.")
						atomic.AddInt32(&totalErrors, 1)
					} else {
						totalRefreshed += int(putIdx)
					}
					return nil
				}
			}

			if deleteIdx > 0 {
				taskC <- func() error {
					if err := datastore.Delete(c, deleteBuf[:deleteIdx]); err != nil {
						log.Fields{
							log.ErrorKey: err,
							"size":       deleteIdx,
						}.Errorf(c, "Failed to Delete batch.")
						atomic.AddInt32(&totalErrors, 1)
					} else {
						totalPruned += int(deleteIdx)
					}
					return nil
				}
			}
		})
		return nil
	}

	err := datastore.RunBatch(c, ms.queryBatchSize, q, func(e *entry) error {
		totalEntries++
		ms.observeEntry()
		entries = append(entries, e)

		if len(entries) >= int(ms.queryBatchSize) {
			// Hit the end of a query batch. Process entries.
			handleEntries(c)
		}
		return nil
	})
	if err != nil {
		return errors.Annotate(err, "failed to run entry query").Err()
	}

	// Flush any outstanding entries (ignore error, will always be nil).
	_ = handleEntries(c)
	if totalErrors > 0 {
		ms.observeErrors(totalErrors)
	}

	log.Fields{
		"entries":   totalEntries,
		"errors":    totalErrors,
		"refreshed": totalRefreshed,
		"pruned":    totalPruned,
	}.Infof(c, "Successfully updated cache entries.")
	return nil
}
581465
func (n *nsLockMap) RUnlock(volume, path, opsID string) {
	readLock := true
	n.unlock(volume, path, opsID, readLock)
}
267335
func (cache *DiskQuotaCacheLocal) getQuotaLocked(
	id keybase1.UserOrTeamID, metered bool) (
	info kbfsblock.QuotaInfo, err error) {
	var hitMeter, missMeter *CountMeter
	if metered {
		hitMeter = cache.hitMeter
		missMeter = cache.missMeter
	}

	quotaBytes, err := cache.db.GetWithMeter(
		[]byte(id.String()), hitMeter, missMeter)
	if err != nil {
		return kbfsblock.QuotaInfo{}, err
	}
	err = cache.config.Codec().Decode(quotaBytes, &info)
	if err != nil {
		return kbfsblock.QuotaInfo{}, err
	}
	return info, nil
}
1212241
func (c *Command) Runnable() bool {
	return c.Run != nil || c.RunE != nil
}
216096
func (l *lockfile) Locked() bool {
	l.stateMutex.Lock()
	defer l.stateMutex.Unlock()
	return l.locked && (l.locktype == unix.F_WRLCK)
}
828538
func (m *Mutex) Lock() {
	spin := NewSpinner(m.priority)
	for !m.TryLock() {
		spin.Yield()
	}
}
344311
func (h *HostStatsCollector) collectLocked() error {
	hs := &HostStats{Timestamp: time.Now().UTC().UnixNano()}

	// Determine up-time
	uptime, err := host.Uptime()
	if err != nil {
		return err
	}
	hs.Uptime = uptime

	// Collect memory stats
	mstats, err := h.collectMemoryStats()
	if err != nil {
		return err
	}
	hs.Memory = mstats

	// Collect cpu stats
	cpus, ticks, err := h.collectCPUStats()
	if err != nil {
		return err
	}
	hs.CPU = cpus
	hs.CPUTicksConsumed = ticks

	// Collect disk stats
	diskStats, err := h.collectDiskStats()
	if err != nil {
		return err
	}
	hs.DiskStats = diskStats

	// Getting the disk stats for the allocation directory
	usage, err := disk.Usage(h.allocDir)
	if err != nil {
		return fmt.Errorf("failed to find disk usage of alloc_dir %q: %v", h.allocDir, err)
	}
	hs.AllocDirStats = h.toDiskStats(usage, nil)

	// Collect devices stats
	deviceStats := h.collectDeviceGroupStats()
	hs.DeviceStats = deviceStats

	// Update the collected status object.
	h.hostStats = hs

	return nil
}
596039
func (m *RWMutex) IsDeadlocked(timeout time.Duration) (o bool, c string) {
	o = true
	c = m.lastSuccessfulLockCaller
	var channelLockAcquired = make(chan bool)
	go func() {
		m.mutex.Lock()
		defer m.mutex.Unlock()
		close(channelLockAcquired)
	}()
	for {
		select {
		case <-channelLockAcquired:
			o = false
			return
		case <-time.After(timeout):
			return
		}
	}
	return
}
1033834
func (sm *schedulerMetrics) FinishRun(tid string, succeeded bool) {
	status := statusString(succeeded)

	sm.totalRunsActive.Dec()
	sm.totalRunsComplete.WithLabelValues(status).Inc()

	sm.runsActive.WithLabelValues(tid).Dec()
	sm.runsComplete.WithLabelValues(tid, status).Inc()
}
965790
func (rw *BoundedRWLock) RUnlock() {
	rw.control.Lock()
	rw.readers--
	if rw.readers == 0 {
		rw.processQueue()
	}
	rw.control.Unlock()
}
3587
func (m *RWMutex) Lock() {
	atomic.AddInt64(&m.pendingWriters, 1)
	m.mutex.Lock()
	atomic.AddInt64(&m.writers, 1)
	atomic.AddInt64(&m.pendingWriters, -1)

	stackBufLen := RWMutexStackBufferLength()

	m.stateMutex.Lock()
	if len(m.stackBuf) < stackBufLen {
		m.stackBuf = make([]byte, stackBufLen)
	}
	n := runtime.Stack(m.stackBuf, false)
	m.lastLockStack = m.stackBuf[:n]
	m.stateMutex.Unlock()
}
198151
func (g *GlobalContext) shutdownCachesLocked() {

	// shutdown and nil out any existing caches.
	if g.trackCache != nil {
		g.trackCache.Shutdown()
	}
	if g.identify2Cache != nil {
		g.identify2Cache.Shutdown()
	}
	if g.linkCache != nil {
		g.linkCache.Shutdown()
	}
	if g.cardCache != nil {
		g.cardCache.Shutdown()
	}
}
295072
func (state *LockState) currLocked() *exclusionState {
	stateCount := len(state.exclusionStates)
	if stateCount == 0 {
		return nil
	}
	return &state.exclusionStates[stateCount-1]
}
316725
func (rw LeveledRWMutex) AssertUnlocked(lockState *LockState) {
	et := lockState.getExclusionType(rw.level)
	if et != nonExclusion {
		panic(unexpectedExclusionError{
			levelToString: lockState.levelToString,
			level:         rw.level,
			exclusionType: et,
		})
	}
}
488928
func (lobby *Lobby) Lock() {
	mu.RLock()
	lock, ok := lobbyLocks[lobby.ID]
	mu.RUnlock()
	if ok {
		lock.Lock()
	}
}
599574
func (m *RWMutex) RUnlock() {
	m.mutex.RUnlock()
	if m.log {
		astilog.Debugf("RUnlock executed for %s", m.name)
	}
}
945284
func (cmd *Cmd) RunBlock(name string, body []string, args []string, newscope bool) (stop bool) {
	if args != nil {
		args = append([]string{name}, args...)
	}

	prev := cmd.context.ScanBlock(body)
	if newscope {
		cmd.context.PushScope(nil, args)
	}
	shouldStop := cmd.runLoop(false)
	if newscope {
		cmd.context.PopScope()
	}
	cmd.context.SetScanner(prev)

	if name == "" { // if stop is called in an unamed block (i.e. not a function) we should really stop
		stop = shouldStop
	}

	return
}