// this function looks for an entirely zero'd work item and returns true. This lets the client
// control how often the batches get flushed (and it can end a request with a null byte)
356074
func zeroWorkItem(data []byte, workItemSize int) bool {
	if len(data)%workItemSize != 0 {
		data = data[:len(data)-(len(data)%workItemSize)]
	}
	dataLen := len(data)
	for i := 0; i < dataLen; i += workItemSize {
		j := 0
		for j < workItemSize && data[i+j] == 0 {
			j++
		}
		if j == workItemSize {
			return true
		}
	}
	return false
}
441232
func (c *FakeJobController) IsCleaningJobRunning(pvName string) bool {
	c.IsRunningCount++
	_, exists := c.pvCleanupRunning[pvName]
	return exists
}
1336265
func (c *WorkItemTypeCache) Get(id uuid.UUID) (WorkItemType, bool) {
	c.mapLock.RLock()
	defer c.mapLock.RUnlock()
	w, ok := c.cache[id]
	return w, ok
}
904689
func (bwa *BatchedWriterAssembly) Flush() {
	if bwa.writer != nil {
		bwa.assembly.SetWriter(bwa.writer)
		bwa.Batch.Flush(bwa.assembly.Write)
	} else {
		bwa.Batch.Flush(bwa.assembly.Flush)
	}
}
1286752
func (c *AggregationController) processNextWorkItem() bool {
	key, quit := c.queue.Get()
	defer c.queue.Done(key)
	if quit {
		return false
	}

	if aggregator.IsLocalAPIService(key.(string)) {
		// for local delegation targets that are aggregated once per second, log at
		// higher level to avoid flooding the log
		klog.V(5).Infof("OpenAPI AggregationController: Processing item %s", key)
	} else {
		klog.Infof("OpenAPI AggregationController: Processing item %s", key)
	}

	action, err := c.syncHandler(key.(string))
	if err == nil {
		c.queue.Forget(key)
	} else {
		utilruntime.HandleError(fmt.Errorf("loading OpenAPI spec for %q failed with: %v", key, err))
	}

	switch action {
	case syncRequeue:
		if aggregator.IsLocalAPIService(key.(string)) {
			klog.V(7).Infof("OpenAPI AggregationController: action for local item %s: Requeue after %s.", key, successfulUpdateDelayLocal)
			c.queue.AddAfter(key, successfulUpdateDelayLocal)
		} else {
			klog.V(7).Infof("OpenAPI AggregationController: action for item %s: Requeue.", key)
			c.queue.AddAfter(key, successfulUpdateDelay)
		}
	case syncRequeueRateLimited:
		klog.Infof("OpenAPI AggregationController: action for item %s: Rate Limited Requeue.", key)
		c.queue.AddRateLimited(key)
	case syncNothing:
		klog.Infof("OpenAPI AggregationController: action for item %s: Nothing (removed from the queue).", key)
	}

	return true
}
161144
func (q *TimedWorkerQueue) CancelWork(key string) bool {
	q.Lock()
	defer q.Unlock()
	worker, found := q.workers[key]
	result := false
	if found {
		klog.V(4).Infof("Cancelling TimedWorkerQueue item %v at %v", key, time.Now())
		if worker != nil {
			result = true
			worker.Cancel()
		}
		delete(q.workers, key)
	}
	return result
}
154032
func (b *basicBatcher) Get() ([]interface{}, error) {
	// Don't check disposed yet so any items remaining in the queue
	// will be returned properly.

	var timeout <-chan time.Time
	if b.maxTime > 0 {
		timeout = time.After(b.maxTime)
	}

	select {
	case items, ok := <-b.batchChan:
		// If there's something on the batch channel, we definitely want that.
		if !ok {
			return nil, ErrDisposed
		}
		return items, nil
	case <-timeout:
		// It's possible something was added to the channel after something
		// was received on the timeout channel, in which case that must
		// be returned first to satisfy our ordering guarantees.
		// We can't just grab the lock here in case the batch channel is full,
		// in which case a Put or Flush will be blocked and holding
		// onto the lock. In that case, there should be something on the
		// batch channel
		for {
			if b.lock.TryLock() {
				// We have a lock, try to read from channel first in case
				// something snuck in
				select {
				case items, ok := <-b.batchChan:
					b.lock.Unlock()
					if !ok {
						return nil, ErrDisposed
					}
					return items, nil
				default:
				}

				// If that is unsuccessful, nothing was added to the channel,
				// and the temp buffer can't have changed because of the lock,
				// so grab that
				items := b.items
				b.items = make([]interface{}, 0, b.maxItems)
				b.availableBytes = 0
				b.lock.Unlock()
				return items, nil
			} else {
				// If we didn't get a lock, there are two cases:
				// 1) The batch chan is full.
				// 2) A Put or Flush temporarily has the lock.
				// In either case, trying to read something off the batch chan,
				// and going back to trying to get a lock if unsuccessful
				// works.
				select {
				case items, ok := <-b.batchChan:
					if !ok {
						return nil, ErrDisposed
					}
					return items, nil
				default:
				}
			}
		}
	}
}
1132526
func WorkParallel(db *DbGorp, tasks []func(worker *DbWorker), returnResults bool, maxNumWorkers int, timeouts int) (results []interface{}, err error) {
	if maxNumWorkers == 0 {
		maxNumWorkers = len(tasks)
	}

	// Create a container with no status callback
	container := NewDbWorker(db,
		MakeCallback(nil,
			func(value interface{}, worker *DbWorker) {
				task := value.(func(worker *DbWorker))
				task(worker)
			}), maxNumWorkers)
	err = container.Start()
	if err != nil {
		return
	}
	for _, task := range tasks {
		container.InputChannel <- task
	}
	if returnResults {
		for range tasks {
			result := <-container.OutputChannel
			results = append(results, result)
		}
	}

	container.Close(timeouts)
	return
}
148181
func (prod *BatchedProducer) flushBatchOnTimeOut() {
	if prod.Batch.ReachedTimeThreshold(prod.batchTimeout) || prod.Batch.ReachedSizeThreshold(prod.batchFlushCount) {
		prod.flushBatch()
	}
}
574792
func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.workqueue.Get()

	if shutdown {
		return false
	}

	// We wrap this block in a func so we can defer c.workqueue.Done.
	err := func(obj interface{}) error {
		// We call Done here so the workqueue knows we have finished
		// processing this item. We also must remember to call Forget if we
		// do not want this work item being re-queued. For example, we do
		// not call Forget if a transient error occurs, instead the item is
		// put back on the workqueue and attempted again after a back-off
		// period.
		defer c.workqueue.Done(obj)
		var key string
		var ok bool
		// We expect strings to come off the workqueue. These are of the
		// form namespace/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 that 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 here else we'd go into a loop of attempting to
			// process a work item that is invalid.
			c.workqueue.Forget(obj)
			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}
		// Run the syncHandler, passing it the namespace/name string of the
		// Foo resource to be synced.
		if err := c.syncHandler(key); err != nil {
			// Put the item back on the workqueue to handle any transient errors.
			c.workqueue.AddRateLimited(key)
			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
		}
		// Finally, if no error occurs we Forget this item so it does not
		// get queued again until another change happens.
		c.workqueue.Forget(obj)
		klog.Infof("Successfully synced '%s'", key)
		return nil
	}(obj)

	if err != nil {
		utilruntime.HandleError(err)
		return true
	}

	return true
}
477986
func (b *Batch) QueryRowResults() *Row {
	rows, _ := b.QueryResults()
	return (*Row)(rows)

}
71992
func (r Billing_Invoice) GetZeroFeeItemCounts() (resp []datatypes.Container_Product_Item_Category_ZeroFee_Count, err error) {
	err = r.Session.DoRequest("SoftLayer_Billing_Invoice", "getZeroFeeItemCounts", nil, &r.Options, &resp)
	return
}
920831
func (c *Controller) GetDefault() (Worker, error) {
	if c.defaultID == "" {
		return nil, errors.Errorf("no default worker")
	}
	return c.Get(c.defaultID)
}
693857
func (c *Client) IsFileLockedByCurrentCommitter(path string) bool {
	filter := map[string]string{"path": path}
	locks, err := c.searchLocalLocks(filter, 1)
	if err != nil {
		tracerx.Printf("Error searching cached locks: %s\nForcing remote search", err)
		locks, _ = c.searchRemoteLocks(filter, 1)
	}
	return len(locks) > 0
}
801324
func AnyJobNeedsState(jobs ...MachineJob) bool {
	for _, j := range jobs {
		if j.NeedsState() {
			return true
		}
	}
	return false
}
1070950
func isBatch(raw json.RawMessage) bool {
	for _, c := range raw {
		// skip insignificant whitespace (http://www.ietf.org/rfc/rfc4627.txt)
		if c == 0x20 || c == 0x09 || c == 0x0a || c == 0x0d {
			continue
		}
		return c == '['
	}
	return false
}
1315385
func (r *GormWorkItemTypeRepository) CheckExists(ctx context.Context, id uuid.UUID) error {
	defer goa.MeasureSince([]string{"goa", "db", "workitemtype", "exists"}, time.Now())
	log.Info(ctx, map[string]interface{}{
		"wit_id": id,
	}, "Checking if work item type exists")

	_, exists := cache.Get(id)
	if exists {
		return nil
	}
	return repository.CheckExists(ctx, r.db, WorkItemType{}.TableName(), id)
}