// StopWatchPartition is an alias for StopConsumingPartition method. The alias was added in order to conform to messaging.Mux interface.
555542
func (conn *ProtoConnectionFields) StopWatchPartition(topic string, partition int32, offset int64) error {
	return conn.StopConsumingPartition(topic, partition, offset)
}
1007907
func (b *Broker) StopConsuming() {
	b.Broker.StopConsuming()

	b.stopReceiving()

	// Waiting for any tasks being processed to finish
	b.processingWG.Wait()

	// Waiting for the receiving goroutine to have stopped
	b.receivingWG.Wait()
}
831163
func (p *ParallelManager) monitorProgress() {
	go func() {
		ticker := time.NewTicker(monitorPeriod)
		defer ticker.Stop()

		var prevSentBytes, maxBandwidth int64
		var retry int

		for {
			select {
			case <-p.stopMonitorCh:
				// Ordered to quit immediately
				return
			case <-ticker.C:
				// Compute new bandwidth from counted sent bytes
				sentBytes := atomic.LoadInt64(&p.sentBytes)
				bandwidth := sentBytes - prevSentBytes
				prevSentBytes = sentBytes

				if bandwidth <= maxBandwidth {
					retry++
					// We still want to add more workers
					// until we are sure that it is not
					// useful to add more of them.
					if retry > 2 {
						return
					}
				} else {
					retry = 0
					maxBandwidth = bandwidth
				}

				for i := 0; i < defaultWorkerFactor; i++ {
					p.addWorker()
				}
			}
		}
	}()
}
617855
func (c *C) StopTimer() {
	if c.timerOn {
		c.duration += time.Now().Sub(c.start)
		c.timerOn = false
		runtime.ReadMemStats(&memStats)
		c.netAllocs += memStats.Mallocs - c.startAllocs
		c.netBytes += memStats.TotalAlloc - c.startBytes
	}
}
1050895
func (c *FakeCiliumEndpoints) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(ciliumendpointsResource, c.ns, opts))

}
1129019
func NewStopwatch(name, vCenter string) *Stopwatch {
	return &Stopwatch{
		stat:  selfstat.RegisterTiming("vsphere", name+"_ns", map[string]string{"vcenter": vCenter}),
		start: time.Now(),
	}
}
689782
func (b *Broker) AddPartitionsToTxn(request *AddPartitionsToTxnRequest) (*AddPartitionsToTxnResponse, error) {
	response := new(AddPartitionsToTxnResponse)

	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}

	return response, nil
}
552915
func (c *Clock) StopAt(now time.Time) {
	c.observe("total", now.Sub(c.first))
}
945200
func (s *AwsSecurityFindingFilters) SetResourcePartition(v []*StringFilter) *AwsSecurityFindingFilters {
	s.ResourcePartition = v
	return s
}
313361
func NewPartition(sfile *tsdb.SeriesFile, path string) *Partition {
	partition := &Partition{
		path:        path,
		sfile:       sfile,
		seriesIDSet: tsdb.NewSeriesIDSet(),

		MaxLogFileSize: DefaultMaxIndexLogFileSize,

		logger:  zap.NewNop(),
		version: Version,
	}

	defaultLabels := prometheus.Labels{"index_partition": ""}
	partition.tracker = newPartitionTracker(newPartitionMetrics(nil), defaultLabels)
	return partition
}
1191072
func (c *SampleController) StartWatch(namespace string, stopCh chan struct{}) error {

	resourceHandlers := cache.ResourceEventHandlerFuncs{
		AddFunc:    c.onAdd,
		UpdateFunc: c.onUpdate,
		DeleteFunc: c.onDelete,
	}
	restClient := c.sampleClientset.RESTClient()
	watcher := opkit.NewWatcher(sample.SampleResource, namespace, resourceHandlers, restClient)
	go watcher.Watch(&sample.Sample{}, stopCh)
	return nil
}
620620
func NewPartitionConsumer(client Client, config *ConsumerConfig, topic string, partition int32, strategy Strategy) PartitionConsumer {
	var metrics PartitionConsumerMetrics = noOpPartitionConsumerMetrics
	if config.EnableMetrics {
		metrics = NewKafkaPartitionConsumerMetrics(topic, partition)
	}

	return &KafkaPartitionConsumer{
		client:    client,
		config:    config,
		topic:     topic,
		partition: partition,
		strategy:  strategy,
		metrics:   metrics,
		stop:      make(chan struct{}, 1),
	}
}
1345089
func (w *Watcher) Stop() {
	w.stopOnce.Do(func() {
		close(w.stopWatch)
		log.WithField(fieldWatcher, w).Debug("Stopped watcher")
		w.stopWait.Wait()
	})
}
613682
func (e *RadixHashJoinExec) doInnerPartition(workerID int) {
	chkIdx, chkNum := workerID, e.innerResult.NumChunks()
	for ; chkIdx < chkNum; chkIdx += int(e.concurrency) {
		if e.finished.Load().(bool) {
			return
		}
		chk := e.innerResult.GetChunk(chkIdx)
		for srcRowIdx, partPtr := range e.innerRowPrts[chkIdx] {
			if partPtr == partPtr4NullKey {
				continue
			}
			partIdx, destRowIdx := partPtr.partitionIdx, partPtr.rowIdx
			part := e.innerParts[partIdx]
			part.Insert(int(destRowIdx), chk.GetRow(srcRowIdx))
		}
	}
}
53257
func (c *KafkaConsumer) Stop() {
	for topic, partitions := range c.Assignment() {
		for _, partition := range partitions {
			c.Remove(topic, partition)
		}
	}
	c.metrics.Stop()
	close(c.stopped)
}
301058
func (wc *watchChan) processEvent(wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		select {
		case e := <-wc.incomingEventChan:
			res := wc.transform(e)
			if res == nil {
				continue
			}
			if len(wc.resultChan) == outgoingBufSize {
				klog.V(3).Infof("Fast watcher, slow processing. Number of buffered events: %d."+
					"Probably caused by slow dispatching events to watchers", outgoingBufSize)
			}
			// If user couldn't receive results fast enough, we also block incoming events from watcher.
			// Because storing events in local will cause more memory usage.
			// The worst case would be closing the fast watcher.
			select {
			case wc.resultChan <- *res:
			case <-wc.ctx.Done():
				return
			}
		case <-wc.ctx.Done():
			return
		}
	}
}
495904
func (f *folder) stopWatch() {
	f.watchMut.Lock()
	f.watchCancel()
	prevErr := f.watchErr
	f.watchErr = errWatchNotStarted
	f.watchMut.Unlock()
	if prevErr != errWatchNotStarted {
		data := map[string]interface{}{
			"folder": f.ID,
			"to":     errWatchNotStarted.Error(),
		}
		if prevErr != nil {
			data["from"] = prevErr.Error()
		}
		events.Default.Log(events.FolderWatchStateChanged, data)
	}
}