// MarkAsProcessed marks the provided offset as highest processed offset if
// it's higher than any previous offset it has received.
197237
func (pot *partitionOffsetTracker) markAsProcessed(offset int64) bool {
	pot.l.Lock()
	defer pot.l.Unlock()
	if offset > pot.highestProcessedOffset {
		pot.highestProcessedOffset = offset
		if pot.waitingForOffset == pot.highestProcessedOffset {
			close(pot.done)
		}
		return true
	} else {
		return false
	}
}
1358993
func (s *ExportJobResponse) SetTotalProcessed(v int64) *ExportJobResponse {
	s.TotalProcessed = &v
	return s
}
544356
func (p *prober) markIPsLocked() {
	for ip, node := range p.nodes {
		node.deletionMark = true
		p.nodes[ip] = node
	}
}
71433
func (s *BulkDeploymentMetrics) SetRecordsProcessed(v int64) *BulkDeploymentMetrics {
	s.RecordsProcessed = &v
	return s
}
732842
func (prc *CorrectionProcessor) Process(nim image.Image, params imageserver.Params) (image.Image, error) {
	enabled, err := prc.isEnabled(params)
	if err != nil {
		return nil, err
	}
	if enabled {
		return prc.process(nim, params)
	}
	return prc.Processor.Process(nim, params)
}
1209857
func (ri *readerOffsetsIterator) Done() {
	if !ri.del {
		return
	}
	ri.del = false

	j, psub, pi := 0, 0, 0
	for i, v := range ri.r.offsets {
		for pi < len(ri.r.prefixes) && i >= ri.r.prefixes[pi].total {
			ri.r.prefixes[pi].total -= psub
			pi++
		}

		if v&(1<<31) > 0 {
			psub++
			continue
		}

		if i != j {
			ri.r.offsets[j] = ri.r.offsets[i]
		}
		j++
	}

	ri.r.offsets = ri.r.offsets[:j]
}
420699
func (e *edge) processUpdate(upt pipe.Receiver) (depChanged bool) {
	// response for cachemap request
	if upt == e.cacheMapReq && upt.Status().Completed {
		if err := upt.Status().Err; err != nil {
			e.cacheMapReq = nil
			if !upt.Status().Canceled && e.err == nil {
				e.err = err
			}
		} else {
			resp := upt.Status().Value.(*cacheMapResp)
			e.cacheMap = resp.CacheMap
			e.cacheMapDone = resp.complete
			e.cacheMapIndex++
			if len(e.deps) == 0 {
				e.cacheMapDigests = append(e.cacheMapDigests, e.cacheMap.Digest)
				if !e.op.IgnoreCache() {
					keys, err := e.op.Cache().Query(nil, 0, e.cacheMap.Digest, e.edge.Index)
					if err != nil {
						logrus.Error(errors.Wrap(err, "invalid query response")) // make the build fail for this error
					} else {
						for _, k := range keys {
							records, err := e.op.Cache().Records(k)
							if err != nil {
								logrus.Errorf("error receiving cache records: %v", err)
								continue
							}

							for _, r := range records {
								e.cacheRecords[r.ID] = r
							}

							e.keys = append(e.keys, e.makeExportable(k, records))
						}
					}
				}
				e.state = edgeStatusCacheSlow
			}
			if e.allDepsHaveKeys(false) {
				e.keysDidChange = true
			}
			// probe keys that were loaded before cache map
			for i, dep := range e.deps {
				e.probeCache(dep, withSelector(dep.keys, e.cacheMap.Deps[i].Selector))
				e.checkDepMatchPossible(dep)
			}
			if !e.cacheMapDone {
				e.cacheMapReq = nil
			}
		}
		return true
	}

	// response for exec request
	if upt == e.execReq && upt.Status().Completed {
		if err := upt.Status().Err; err != nil {
			e.execReq = nil
			if !upt.Status().Canceled && e.err == nil {
				e.err = err
			}
		} else {
			e.result = NewSharedCachedResult(upt.Status().Value.(CachedResult))
			e.state = edgeStatusComplete
		}
		return true
	}

	// response for requests to dependencies
	if dep, ok := e.depRequests[upt]; ok {
		if err := upt.Status().Err; !upt.Status().Canceled && upt.Status().Completed && err != nil {
			if e.err == nil {
				e.err = err
			}
			dep.err = err
		}

		state := upt.Status().Value.(*edgeState)

		if len(dep.keys) < len(state.keys) {
			newKeys := state.keys[len(dep.keys):]
			if e.cacheMap != nil {
				e.probeCache(dep, withSelector(newKeys, e.cacheMap.Deps[dep.index].Selector))
				dep.edgeState.keys = state.keys
				if e.allDepsHaveKeys(false) {
					e.keysDidChange = true
				}
			}
			depChanged = true
		}
		if dep.state != edgeStatusComplete && state.state == edgeStatusComplete {
			e.keysDidChange = true
		}

		recheck := state.state != dep.state

		dep.edgeState = *state

		if recheck && e.cacheMap != nil {
			e.checkDepMatchPossible(dep)
			depChanged = true
		}

		return
	}

	// response for result based cache function
	for i, dep := range e.deps {
		if upt == dep.slowCacheReq && upt.Status().Completed {
			if err := upt.Status().Err; err != nil {
				dep.slowCacheReq = nil
				if !upt.Status().Canceled && e.err == nil {
					e.err = upt.Status().Err
				}
			} else if !dep.slowCacheComplete {
				k := NewCacheKey(upt.Status().Value.(digest.Digest), -1)
				dep.slowCacheKey = &ExportableCacheKey{CacheKey: k, Exporter: &exporter{k: k}}
				slowKeyExp := CacheKeyWithSelector{CacheKey: *dep.slowCacheKey}
				defKeys := make([]CacheKeyWithSelector, 0, len(dep.result.CacheKeys()))
				for _, dk := range dep.result.CacheKeys() {
					defKeys = append(defKeys, CacheKeyWithSelector{CacheKey: dk, Selector: e.cacheMap.Deps[i].Selector})
				}
				dep.slowCacheFoundKey = e.probeCache(dep, []CacheKeyWithSelector{slowKeyExp})

				// connect def key to slow key
				e.op.Cache().Query(append(defKeys, slowKeyExp), dep.index, e.cacheMap.Digest, e.edge.Index)

				dep.slowCacheComplete = true
				e.keysDidChange = true
				e.checkDepMatchPossible(dep) // not matching key here doesn't set nocachematch possible to true
			}
			return true
		}
	}

	return
}
525723
func (shardSwap *shardSchemaSwap) markPropagationDone(err *error) {
	if *err != nil {
		msg := fmt.Sprintf("Error: %v", *err)
		shardSwap.addPropagationLog(msg)
		shardSwap.propagationUINode.Message = msg
		shardSwap.propagationUINode.State = workflowpb.WorkflowState_Done
	} else if shardSwap.numTabletsSwapped == shardSwap.numTabletsTotal {
		shardSwap.propagationUINode.State = workflowpb.WorkflowState_Done
	}
	shardSwap.propagationUINode.BroadcastChanges(false /* updateChildren */)
}
378500
func (a *DateHistogramAggregation) Offset(offset string) *DateHistogramAggregation {
	a.offset = offset
	return a
}
116299
func (c *Container) ProcessAdd(process rproc.Process) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.processes[process.ID()] = process
}
1265981
func (s *DomainStatus) SetProcessing(v bool) *DomainStatus {
	s.Processing = &v
	return s
}
144310
func (s *GroupsService) Mark(channel string, timestamp string) *GroupsMarkCall {
	var call GroupsMarkCall
	call.service = s
	call.channel = channel
	call.timestamp = timestamp
	return &call
}
964121
func MarkTaskSucceeded(t *automationpb.Task, output string) {
	t.State = automationpb.TaskState_DONE
	t.Output = output
}
753343
func (vsdw *VerticalSplitDiffWorker) markAsWillFail(er concurrency.ErrorRecorder, err error) {
	er.RecordError(err)
	vsdw.SetState(WorkerStateDiffWillFail)
}
1267581
func (s *StanServer) processReplicatedSendAndAck(ssa *spb.SubSentAndAck) {
	c, err := s.lookupOrCreateChannel(ssa.Channel)
	if err != nil {
		return
	}
	sub := c.ss.LookupByAckInbox(ssa.AckInbox)
	if sub == nil {
		return
	}
	sub.Lock()
	defer sub.Unlock()

	// This is not optimized. The leader sent all accumulated sent and ack
	// sequences. For queue members, there is no much that can be done
	// because by nature seq will not be contiguous, but for non queue
	// subs, this could be optimized.
	for _, sequence := range ssa.Sent {
		// Update LastSent if applicable
		if sequence > sub.LastSent {
			sub.LastSent = sequence
		}
		// In case this is a queue member, update queue state's LastSent.
		if sub.qstate != nil && sequence > sub.qstate.lastSent {
			sub.qstate.lastSent = sequence
		}
		// Set 0 for expiration time. This will be computed
		// when the follower becomes leader and attempts to
		// redeliver messages.
		sub.acksPending[sequence] = 0
	}
	// Now remove the acks pending that we potentially just added ;-)
	for _, sequence := range ssa.Ack {
		delete(sub.acksPending, sequence)
	}
	// Don't set the sub.stalled here. Let that be done if the server
	// becomes leader and attempt the first deliveries.
}
721110
func (m *Metrics) Finalize(rateLimit uint, totalQueries uint) {
	if rateLimit == 0 {
		rateLimit = 1
	}

	for _, s := range m.TimeMetrics {
		sort.Float64s(s.vals)
		cnt := len(s.vals)

		s.Min = Float64(s.vals[0])
		s.Med = Float64(s.vals[(50*cnt)/100]) // median = 50th percentile
		s.P95 = Float64(s.vals[(95*cnt)/100])
		s.Max = Float64(s.vals[cnt-1])
		s.Sum = (s.Sum * float64(rateLimit)) + s.outlierSum
		s.Avg = Float64(s.Sum / float64(totalQueries))
	}

	for _, s := range m.NumberMetrics {
		sort.Sort(byUint64(s.vals))
		cnt := len(s.vals)

		s.Min = Uint64(s.vals[0])
		s.Med = Uint64(s.vals[(50*cnt)/100]) // median = 50th percentile
		s.P95 = Uint64(s.vals[(95*cnt)/100])
		s.Max = Uint64(s.vals[cnt-1])
		s.Sum = (s.Sum * uint64(rateLimit)) + s.outlierSum
		s.Avg = Uint64(s.Sum / uint64(totalQueries))
	}

	for _, s := range m.BoolMetrics {
		s.Sum = (s.Sum * uint64(rateLimit)) + s.outlierSum
	}
}
1129930
func (w *WalletState) MarkAsRead(ctx context.Context, accountID stellar1.AccountID, mostRecentID stellar1.TransactionID) error {
	err := w.Remoter.MarkAsRead(ctx, accountID, mostRecentID)
	if err == nil {
		mctx := libkb.NewMetaContext(ctx, w.G())
		if rerr := w.RefreshAsync(mctx, accountID, "MarkAsRead"); rerr != nil {
			mctx.Debug("Refresh after MarkAsRead error: %s", err)
		}
	}
	return err
}