// ProcessQSizeOption sets the processor queue size(length) and
// returns the previous queue option state.
324844
func ProcessQSizeOption(v uint) workManagerOption {
	return func(w *workManager) workManagerOption {
		previous := w.processQSize
		w.processQSize = v
		return ProcessQSizeOption(previous)
	}
}
779227
func (c *Config) ProcessOptions(options ...Option) {
	for _, opt := range options {
		if opt != nil {
			opt(c)
		}
	}
}
452289
func WithSize(size string) ProviderOpts {
	return func(p *Provider) error {
		p.Size = size
		return nil
	}
}
965658
func (engine *Engine) BufferSize(size int) *Session {
	session := engine.NewSession()
	session.isAutoClose = true
	return session.BufferSize(size)
}
1129450
func (kcp *KCP) PeekSize() (length int) {
	if len(kcp.rcv_queue) == 0 {
		return -1
	}

	seg := &kcp.rcv_queue[0]
	if seg.frg == 0 {
		return len(seg.data)
	}

	if len(kcp.rcv_queue) < int(seg.frg+1) {
		return -1
	}

	for k := range kcp.rcv_queue {
		seg := &kcp.rcv_queue[k]
		length += len(seg.data)
		if seg.frg == 0 {
			break
		}
	}
	return
}
137543
func WithWaiterRequestOptions(opts ...Option) WaiterOption {
	return func(w *Waiter) {
		w.RequestOptions = append(w.RequestOptions, opts...)
	}
}
114940
func getPageSizeMountOptionFromPod(pod *v1.Pod) (string, error) {
	pageSizeFound := false
	pageSize := resource.Quantity{}
	// In some rare cases init containers can also consume Huge pages.
	containers := append(pod.Spec.Containers, pod.Spec.InitContainers...)
	for _, container := range containers {
		// We can take request because limit and requests must match.
		for requestName := range container.Resources.Requests {
			if v1helper.IsHugePageResourceName(requestName) {
				currentPageSize, err := v1helper.HugePageSizeFromResourceName(requestName)
				if err != nil {
					return "", err
				}
				// PageSize for all volumes in a POD are equal, except for the first one discovered.
				if pageSizeFound && pageSize.Cmp(currentPageSize) != 0 {
					return "", fmt.Errorf("multiple pageSizes for huge pages in a single PodSpec")
				}
				pageSize = currentPageSize
				pageSizeFound = true
			}
		}
	}

	if !pageSizeFound {
		return "", fmt.Errorf("hugePages storage requested, but there is no resource request for huge pages")
	}

	return fmt.Sprintf("%s=%s", hugePagesPageSizeMountOption, pageSize.String()), nil

}
290491
func ProcessJSON(input []byte, options *ProcessorOptions) ([]byte, error) {

	// First, unmarshal the JSON to a generic interface{} type
	var unmarshalled interface{}
	if err := json.Unmarshal(input, &unmarshalled); err != nil {
		return nil, fmt.Errorf("invalid JSON: %s", err)
	}

	var processed interface{}

	if options != nil && options.NoProcess {
		processed = unmarshalled
	} else {
		applyGlobals(unmarshalled, options)

		overrideParameters(unmarshalled, options)

		evaluateConditions(unmarshalled, options)

		// Process all of the intrinsic functions
		processed = search(unmarshalled, unmarshalled, options)
	}

	// And return the result back as a []byte of JSON
	result, err := json.MarshalIndent(processed, "", "  ")
	if err != nil {
		return nil, fmt.Errorf("invalid JSON: %s", err)
	}

	return result, nil
}
1127791
func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) {
	idealPoolSize := int(float64(num) * 0.25)
	if idealPoolSize < 1 {
		idealPoolSize = 1
	} else if idealPoolSize > n.getOpts().QueueScanWorkerPoolMax {
		idealPoolSize = n.getOpts().QueueScanWorkerPoolMax
	}
	for {
		if idealPoolSize == n.poolSize {
			break
		} else if idealPoolSize < n.poolSize {
			// contract
			closeCh <- 1
			n.poolSize--
		} else {
			// expand
			n.waitGroup.Wrap(func() {
				n.queueScanWorker(workCh, responseCh, closeCh)
			})
			n.poolSize++
		}
	}
}
427542
func (pq *PriorityQueue) Length() uint64 {
	pq.RLock()
	defer pq.RUnlock()

	var length uint64
	for _, v := range pq.levels {
		length += v.length()
	}

	return length
}
728406
func ListSizeOp(binName string) *Operation {
	return &Operation{opType: _CDT_READ, binName: binName, binValue: ListValue{_CDT_LIST_SIZE}, encoder: listGenericOpEncoder}
}
637239
func PrefetchSize(size int) Option {
	return func(r *Rabbus) error {
		r.config.qos.prefetchSize = size
		return nil
	}
}
1066984
func ReconnectBufSize(size int) Option {
	return func(o *Options) error {
		o.ReconnectBufSize = size
		return nil
	}
}
781674
func (s *service) Init(opts ...Option) {
	// process options
	for _, o := range opts {
		o(&s.opts)
	}

	s.once.Do(func() {
		// Initialise the command flags, overriding new service
		_ = s.opts.Cmd.Init(
			cmd.Broker(&s.opts.Broker),
			cmd.Registry(&s.opts.Registry),
			cmd.Transport(&s.opts.Transport),
			cmd.Client(&s.opts.Client),
			cmd.Server(&s.opts.Server),
		)
	})
}
679776
func (p Packet) ParseOptions() Options {
	opts := p.Options()
	options := make(Options, 10)
	for len(opts) >= 2 && OptionCode(opts[0]) != End {
		if OptionCode(opts[0]) == Pad {
			opts = opts[1:]
			continue
		}
		size := int(opts[1])
		if len(opts) < 2+size {
			break
		}
		options[OptionCode(opts[0])] = opts[2 : 2+size]
		opts = opts[2+size:]
	}
	return options
}
1358583
func (d *Disque) QueueLength(queueName string) (queueLength int, err error) {
	return redis.Int(d.call("QLEN", redis.Args{}.Add(queueName)))
}
803037
func SelectBufferedQueue(redisHost, redisPort, redisPassword string, redisDB int64, name string, bufferSize int) (queue *BufferedQueue, err error) {
	q, err := SelectQueue(redisHost, redisPort, redisPassword, redisDB, name)
	if err != nil {
		return nil, err
	}

	return &BufferedQueue{
		Queue:        q,
		BufferSize:   bufferSize,
		Buffer:       make(chan *Package, bufferSize*2),
		flushStatus:  make(chan chan bool, 1),
		flushCommand: make(chan bool, bufferSize*2),
	}, nil
}