// processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
507775
func (tc *TokenCleaner) processNextWorkItem() bool {
	key, quit := tc.queue.Get()
	if quit {
		return false
	}
	defer tc.queue.Done(key)

	if err := tc.syncFunc(key.(string)); err != nil {
		tc.queue.AddRateLimited(key)
		utilruntime.HandleError(fmt.Errorf("Sync %v failed with : %v", key, err))
		return true
	}

	tc.queue.Forget(key)
	return true
}
96281
func (b *EvalBroker) processWaitingEnqueue(eval *structs.Evaluation) {
	timer := time.AfterFunc(eval.Wait, func() {
		b.enqueueWaiting(eval)
	})
	b.timeWait[eval.ID] = timer
	b.stats.TotalWaiting += 1
}
1051289
func (l *List) RemoveItem(i int) {
	// Adjust pos and selected before removing.
	if l.pos >= len(l.items) {
		l.pos--
	}
	if l.selected == i {
		l.selected = -1
	} else if l.selected > i {
		l.selected--
	}

	// Copy items following i to position i.
	copy(l.items[i:], l.items[i+1:])

	// Shrink items by one.
	l.items[len(l.items)-1] = ""
	l.items = l.items[:len(l.items)-1]

	if l.onSelectionChanged != nil {
		l.onSelectionChanged(l)
	}
}
13619
func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask, ch chan<- *copResponse) ([]*copTask, error) {
	failpoint.Inject("handleTaskOnceError", func(val failpoint.Value) {
		if val.(bool) {
			failpoint.Return(nil, errors.New("mock handleTaskOnce error"))
		}
	})

	sender := NewRegionRequestSender(worker.store.regionCache, worker.store.client)
	req := &tikvrpc.Request{
		Type: task.cmdType,
		Cop: &coprocessor.Request{
			Tp:     worker.req.Tp,
			Data:   worker.req.Data,
			Ranges: task.ranges.toPBRanges(),
		},
		Context: kvrpcpb.Context{
			IsolationLevel: pbIsolationLevel(worker.req.IsolationLevel),
			Priority:       kvPriorityToCommandPri(worker.req.Priority),
			NotFillCache:   worker.req.NotFillCache,
			HandleTime:     true,
			ScanDetail:     true,
		},
	}
	startTime := time.Now()
	resp, rpcCtx, err := sender.SendReqCtx(bo, req, task.region, ReadTimeoutMedium)
	if err != nil {
		return nil, errors.Trace(err)
	}
	// Set task.storeAddr field so its task.String() method have the store address information.
	task.storeAddr = sender.storeAddr
	costTime := time.Since(startTime)
	if costTime > minLogCopTaskTime {
		worker.logTimeCopTask(costTime, task, bo, resp)
	}
	metrics.TiKVCoprocessorHistogram.Observe(costTime.Seconds())

	if task.cmdType == tikvrpc.CmdCopStream {
		return worker.handleCopStreamResult(bo, rpcCtx, resp.CopStream, task, ch)
	}

	// Handles the response for non-streaming copTask.
	return worker.handleCopResponse(bo, rpcCtx, &copResponse{pbResp: resp.Cop}, task, ch, nil)
}
510369
func (s *iniSection) rmItem(key string, flag bool) bool {
	if s == nil {
		return false
	}
	key = ident(flag, key)
	for i, item := range s.Data {
		if item == nil {
			continue
		}
		if ident(flag, item.Key) == key {
			n := len(s.Data) - 1
			skip := 1
			// Remove newline if the key was the last one in the block.
			if (i > 0 && s.Data[i-1] == nil && i < n && s.Data[i+1] == nil) ||
				(i == 0 && n > 0 && s.Data[1] == nil) {
				// Removed key is not the first key and previous and next keys are newlines.
				// Removed key is the first key and next key is a newline.
				skip++
				n--
			}
			copy(s.Data[i:], s.Data[i+skip:])
			s.Data[n] = nil
			s.Data = s.Data[:n]
			return true
		}
	}
	return false
}
1129762
func (c ConfigDataNode) DeleteItem(k string) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	delete(c.table, k)
}
552853
func (c *WorkItemTypeCache) Clear() {
	c.mapLock.Lock()
	defer c.mapLock.Unlock()
	log.Info(nil, nil, "Clearing work item cache")

	c.cache = make(witCacheMap)
}
558133
func (n *NetworkServerAPI) EnqueueMulticastQueueItem(ctx context.Context, req *ns.EnqueueMulticastQueueItemRequest) (*empty.Empty, error) {
	if req.MulticastQueueItem == nil {
		return nil, grpc.Errorf(codes.InvalidArgument, "multicast_queue_item must not be nil")
	}

	var mgID uuid.UUID
	copy(mgID[:], req.MulticastQueueItem.MulticastGroupId)

	qi := storage.MulticastQueueItem{
		MulticastGroupID: mgID,
		FCnt:             req.MulticastQueueItem.FCnt,
		FPort:            uint8(req.MulticastQueueItem.FPort),
		FRMPayload:       req.MulticastQueueItem.FrmPayload,
	}

	err := storage.Transaction(func(tx sqlx.Ext) error {
		return multicast.EnqueueQueueItem(storage.RedisPool(), tx, qi)
	})
	if err != nil {
		return nil, errToRPCError(err)
	}

	return &empty.Empty{}, nil
}
1230292
func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.releaseWorkqueue.Get()
	if shutdown {
		return false
	}

	// wrapping block in a func to defer c.workqueue.Done
	err := func(obj interface{}) error {
		// We call Done here so the workqueue knows we have finished
		// processing this item. We must call Forget if we do not want
		// this work item being re-queued. If a transient error
		// occurs, we do not call Forget. Instead the item is put back
		// on the workqueue and attempted again after a back-off
		// period.
		defer c.releaseWorkqueue.Done(obj)

		var key string
		var ok bool
		// We expect strings to come off the workqueue. These are of the
		// form "namespace/fhr(custom resource) name". We do this as the delayed nature of the
		// workqueue means the items in the informer cache may actually be
		// more up to date than when the item was initially put onto the
		// workqueue.
		if key, ok = obj.(string); !ok {
			// As the item in the workqueue is actually invalid, we call
			// Forget not to get into a loop of attempting to
			// process a work item that is invalid.
			c.releaseWorkqueue.Forget(obj)
			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))

			return nil
		}
		// Run the syncHandler, passing it the namespace/name string of the
		// HelmRelease resource to sync the corresponding Chart release.
		// If the sync failed, then we return while the item will get requeued
		if err := c.syncHandler(key); err != nil {
			return fmt.Errorf("errored syncing HelmRelease '%s': %s", key, err.Error())
		}
		// If no error occurs we Forget this item so it does not
		// get queued again until another change happens.
		c.releaseWorkqueue.Forget(obj)
		return nil
	}(obj)

	if err != nil {
		runtime.HandleError(err)
		return true
	}
	return true
}
671586
func processEventQueue(xu *xgbutil.XUtil, pingBefore, pingAfter chan struct{}) {
	for !Empty(xu) {
		if Quitting(xu) {
			return
		}

		// We send the ping *before* the next event is dequeued.
		// This is so the queue doesn't present a misrepresentation of which
		// events haven't been processed yet.
		if pingBefore != nil && pingAfter != nil {
			pingBefore <- struct{}{}
		}
		ev, err := Dequeue(xu)

		// If we gobbled up an error, send it to the error event handler
		// and move on the next event/error.
		if err != nil {
			ErrorHandlerGet(xu)(err)
			if pingBefore != nil && pingAfter != nil {
				pingAfter <- struct{}{}
			}
			continue
		}

		// We know there isn't an error. If there isn't an event either,
		// then there's a bug somewhere.
		if ev == nil {
			xgbutil.Logger.Fatal("BUG: Expected an event but got nil.")
		}

		hooks := getHooks(xu)
		for _, hook := range hooks {
			if !hook.Run(xu, ev) {
				goto END
			}
		}

		switch event := ev.(type) {
		case xproto.KeyPressEvent:
			e := KeyPressEvent{&event}

			// If we're redirecting key events, this is the place to do it!
			if wid := RedirectKeyGet(xu); wid > 0 {
				e.Event = wid
			}

			xu.TimeSet(e.Time)
			runCallbacks(xu, e, KeyPress, e.Event)
		case xproto.KeyReleaseEvent:
			e := KeyReleaseEvent{&event}

			// If we're redirecting key events, this is the place to do it!
			if wid := RedirectKeyGet(xu); wid > 0 {
				e.Event = wid
			}

			xu.TimeSet(e.Time)
			runCallbacks(xu, e, KeyRelease, e.Event)
		case xproto.ButtonPressEvent:
			e := ButtonPressEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, ButtonPress, e.Event)
		case xproto.ButtonReleaseEvent:
			e := ButtonReleaseEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, ButtonRelease, e.Event)
		case xproto.MotionNotifyEvent:
			e := MotionNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, MotionNotify, e.Event)
		case xproto.EnterNotifyEvent:
			e := EnterNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, EnterNotify, e.Event)
		case xproto.LeaveNotifyEvent:
			e := LeaveNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, LeaveNotify, e.Event)
		case xproto.FocusInEvent:
			e := FocusInEvent{&event}
			runCallbacks(xu, e, FocusIn, e.Event)
		case xproto.FocusOutEvent:
			e := FocusOutEvent{&event}
			runCallbacks(xu, e, FocusOut, e.Event)
		case xproto.KeymapNotifyEvent:
			e := KeymapNotifyEvent{&event}
			runCallbacks(xu, e, KeymapNotify, NoWindow)
		case xproto.ExposeEvent:
			e := ExposeEvent{&event}
			runCallbacks(xu, e, Expose, e.Window)
		case xproto.GraphicsExposureEvent:
			e := GraphicsExposureEvent{&event}
			runCallbacks(xu, e, GraphicsExposure, xproto.Window(e.Drawable))
		case xproto.NoExposureEvent:
			e := NoExposureEvent{&event}
			runCallbacks(xu, e, NoExposure, xproto.Window(e.Drawable))
		case xproto.VisibilityNotifyEvent:
			e := VisibilityNotifyEvent{&event}
			runCallbacks(xu, e, VisibilityNotify, e.Window)
		case xproto.CreateNotifyEvent:
			e := CreateNotifyEvent{&event}
			runCallbacks(xu, e, CreateNotify, e.Parent)
		case xproto.DestroyNotifyEvent:
			e := DestroyNotifyEvent{&event}
			runCallbacks(xu, e, DestroyNotify, e.Window)
		case xproto.UnmapNotifyEvent:
			e := UnmapNotifyEvent{&event}
			runCallbacks(xu, e, UnmapNotify, e.Window)
		case xproto.MapNotifyEvent:
			e := MapNotifyEvent{&event}
			runCallbacks(xu, e, MapNotify, e.Event)
		case xproto.MapRequestEvent:
			e := MapRequestEvent{&event}
			runCallbacks(xu, e, MapRequest, e.Window)
			runCallbacks(xu, e, MapRequest, e.Parent)
		case xproto.ReparentNotifyEvent:
			e := ReparentNotifyEvent{&event}
			runCallbacks(xu, e, ReparentNotify, e.Window)
		case xproto.ConfigureNotifyEvent:
			e := ConfigureNotifyEvent{&event}
			runCallbacks(xu, e, ConfigureNotify, e.Window)
		case xproto.ConfigureRequestEvent:
			e := ConfigureRequestEvent{&event}
			runCallbacks(xu, e, ConfigureRequest, e.Window)
			runCallbacks(xu, e, ConfigureRequest, e.Parent)
		case xproto.GravityNotifyEvent:
			e := GravityNotifyEvent{&event}
			runCallbacks(xu, e, GravityNotify, e.Window)
		case xproto.ResizeRequestEvent:
			e := ResizeRequestEvent{&event}
			runCallbacks(xu, e, ResizeRequest, e.Window)
		case xproto.CirculateNotifyEvent:
			e := CirculateNotifyEvent{&event}
			runCallbacks(xu, e, CirculateNotify, e.Window)
		case xproto.CirculateRequestEvent:
			e := CirculateRequestEvent{&event}
			runCallbacks(xu, e, CirculateRequest, e.Window)
		case xproto.PropertyNotifyEvent:
			e := PropertyNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, PropertyNotify, e.Window)
		case xproto.SelectionClearEvent:
			e := SelectionClearEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, SelectionClear, e.Owner)
		case xproto.SelectionRequestEvent:
			e := SelectionRequestEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, SelectionRequest, e.Requestor)
		case xproto.SelectionNotifyEvent:
			e := SelectionNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, SelectionNotify, e.Requestor)
		case xproto.ColormapNotifyEvent:
			e := ColormapNotifyEvent{&event}
			runCallbacks(xu, e, ColormapNotify, e.Window)
		case xproto.ClientMessageEvent:
			e := ClientMessageEvent{&event}
			runCallbacks(xu, e, ClientMessage, e.Window)
		case xproto.MappingNotifyEvent:
			e := MappingNotifyEvent{&event}
			runCallbacks(xu, e, MappingNotify, NoWindow)
		case shape.NotifyEvent:
			e := ShapeNotifyEvent{&event}
			runCallbacks(xu, e, ShapeNotify, e.AffectedWindow)
		default:
			if event != nil {
				xgbutil.Logger.Printf("ERROR: UNSUPPORTED EVENT TYPE: %T",
					event)
			}
		}

	END:

		if pingBefore != nil && pingAfter != nil {
			pingAfter <- struct{}{}
		}
	}
}
118626
func (s *InMemoryStore) SaveNextBatch(userID, nextBatchToken string) {
	s.NextBatch[userID] = nextBatchToken
}
695550
func (wit WorkItemType) ConvertWorkItemStorageToModel(workItem WorkItemStorage) (*WorkItem, error) {
	result := WorkItem{
		ID:                     workItem.ID,
		Number:                 workItem.Number,
		Type:                   workItem.Type,
		Version:                workItem.Version,
		Fields:                 map[string]interface{}{},
		SpaceID:                workItem.SpaceID,
		relationShipsChangedAt: workItem.RelationShipsChangedAt,
	}

	for name, field := range wit.Fields {
		var err error
		if name == SystemCreatedAt {
			continue
		}
		result.Fields[name], err = field.ConvertFromModel(name, workItem.Fields[name])
		if err != nil {
			return nil, errs.WithStack(err)
		}
		result.Fields[SystemOrder] = workItem.ExecutionOrder
	}

	return &result, nil
}
1251318
func (w *workManager) Work(j job) queuedJob {
	qj := newQueuedJob(j)
	switch j.Type() {
	case collectJobType:
		w.collectq.Event <- qj
	case processJobType:
		w.processq.Event <- qj
	case publishJobType:
		w.publishq.Event <- qj
	}
	return qj
}
1306450
func (woc *wfOperationCtx) requeue() {
	key, err := cache.MetaNamespaceKeyFunc(woc.wf)
	if err != nil {
		woc.log.Errorf("Failed to requeue workflow %s: %v", woc.wf.ObjectMeta.Name, err)
		return
	}
	woc.controller.wfQueue.Add(key)
}
981511
func (c *Controller) processNextWorkItem() bool {
	pvcKey, quit := c.queue.Get()
	if quit {
		return false
	}
	defer c.queue.Done(pvcKey)

	pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(pvcKey.(string))
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("Error parsing PVC key %q: %v", pvcKey, err))
		return true
	}

	err = c.processPVC(pvcNamespace, pvcName)
	if err == nil {
		c.queue.Forget(pvcKey)
		return true
	}

	utilruntime.HandleError(fmt.Errorf("PVC %v failed with : %v", pvcKey, err))
	c.queue.AddRateLimited(pvcKey)

	return true
}
897657
func (m *Memberlist) getNextMessage() (msgHandoff, bool) {
	m.msgQueueLock.Lock()
	defer m.msgQueueLock.Unlock()

	if el := m.highPriorityMsgQueue.Back(); el != nil {
		m.highPriorityMsgQueue.Remove(el)
		msg := el.Value.(msgHandoff)
		return msg, true
	} else if el := m.lowPriorityMsgQueue.Back(); el != nil {
		m.lowPriorityMsgQueue.Remove(el)
		msg := el.Value.(msgHandoff)
		return msg, true
	}
	return msgHandoff{}, false
}
137558
func (deallocator *Deallocator) processService(ctx context.Context, service *api.Service) (bool, error) {
	if !service.PendingDelete {
		return false, nil
	}

	var (
		tasks []*api.Task
		err   error
	)

	deallocator.store.View(func(tx store.ReadTx) {
		tasks, err = store.FindTasks(tx, store.ByServiceID(service.ID))
	})

	if err != nil {
		log.G(ctx).WithError(err).Errorf("failed to retrieve the list of tasks for service %v", service.ID)
		// if in doubt, let's proceed to clean up the service anyway
		// better to clean up resources that shouldn't be cleaned up yet
		// than ending up with a service and some resources lost in limbo forever
		return true, deallocator.deallocateService(ctx, service)
	} else if len(tasks) == 0 {
		// no tasks remaining for this service, we can clean it up
		return true, deallocator.deallocateService(ctx, service)
	}
	deallocator.services[service.ID] = &serviceWithTaskCounts{service: service, taskCount: len(tasks)}
	return false, nil
}