// finishDispatching stops all the watchers that were supposed to be
// stopped in the meantime, but it was deferred to avoid closing input
// channels of watchers, as add() may still have writing to it.
// It also marks dispatching as false.
13342
func (c *Cacher) finishDispatching() {
	c.Lock()
	defer c.Unlock()
	c.dispatching = false
	for _, watcher := range c.watchersToStop {
		watcher.stop()
	}
	c.watchersToStop = c.watchersToStop[:0]
}
1013753
func NewDispatcher(l log.Logger, unknownHandler Handler) Dispatcher {
	return &dispatcher{
		logger:   l,
		unknown:  unknownHandler,
		handlers: make(map[StreamType]Handler, 0),
	}
}
400558
func (mc *MergeControl) mergeFinish(merge *OneMerge) {
	// forceMerge, addIndexes or abortAllmerges may be waiting on
	// merges to finish
	// notifyAll()

	// It's possible we are called twice, eg if there was an error
	// inside mergeInit()
	if merge.registerDone {
		for _, info := range merge.segments {
			delete(mc.mergingSegments, info)
		}
		merge.registerDone = false
	}

	delete(mc.runningMerges, merge)
	if len(mc.runningMerges) == 0 {
		mc.mergeSignal.Signal()
	}
}
500881
func (s *PhasedStopper) StopOutbounds() error {
	if !s.inboundsStopped.Load() {
		return errors.New("must stop inbounds first")
	}
	if s.outboundsStopInitiated.Swap(true) {
		return errors.New("already began stopping outbounds")
	}
	defer s.outboundsStopped.Store(true)
	s.log.Debug("stopping outbounds")
	wait := errorsync.ErrorWaiter{}
	for _, o := range s.dispatcher.outbounds {
		if o.Unary != nil {
			wait.Submit(o.Unary.Stop)
		}
		if o.Oneway != nil {
			wait.Submit(o.Oneway.Stop)
		}
		if o.Stream != nil {
			wait.Submit(o.Stream.Stop)
		}
	}
	if errs := wait.Wait(); len(errs) > 0 {
		return multierr.Combine(errs...)
	}
	s.log.Debug("stopped outbounds")
	return nil
}
132240
func getDispatcher(c context.Context) *tq.Dispatcher {
	return c.Value(&dspKey).(*tq.Dispatcher)
}
650435
func (p *BlockPrefetchClient) Stop() {
	p.Lock()
	close(p.retargetChan)
	p.Unlock()
}
239063
func (w *Watcher) Stop() {
	w.Lock()
	defer w.Unlock()
	if !w.closed {
		notify.Stop(w.evtch)
		close(w.modch)
		w.closed = true
	}
}
917247
func (w *Watcher) Close() error {
	w.exit <- true
	w.consume()
	w.conn.noIdle()

	<-w.done // wait for idle to finish and channels to close
	// At this point, watch goroutine has ended,
	// so it's safe to close connection.
	return w.conn.Close()
}
161711
func (m *Broadcaster) closeAll() {
	m.lock.Lock()
	defer m.lock.Unlock()
	for _, w := range m.watchers {
		close(w.result)
	}
	// Delete everything from the map, since presence/absence in the map is used
	// by stopWatching to avoid double-closing the channel.
	m.watchers = map[int64]*broadcasterWatcher{}
}
1288332
func (l *loop) finalizeTermination() {
	l.status = Stopped
	// Close stopC in case  the termination is due to an
	// error or internal.
	select {
	case <-l.stopC:
	default:
		close(l.stopC)
	}
	// Communicate that it's done.
	select {
	case <-l.doneC:
	default:
		close(l.doneC)
	}
	// If a sentinel monitors us then till him.
	if l.sentinel != nil {
		if l.err != nil {
			// Notify sentinel about error termination.
			l.sentinel.notifyC <- l.owner
		} else {
			// Tell sentinel to remove loop.
			l.sentinel.Forget(l)
		}
	}
	if l.err != nil {
		l.logf(true, "loop %q stopped with error: %v", l, l.err)
	} else {
		l.logf(false, "loop %q stopped", l)
	}
}
276516
func (pw *ProxyWatcher) Stopping() bool {
	pw.mutex.Lock()
	defer pw.mutex.Unlock()
	return pw.stopped
}
592855
func (b *Broker) StopConsuming() {
	// Stop the receiving goroutine
	b.stopReceivingChan <- 1
	// Waiting for the receiving goroutine to have stopped
	b.receivingWG.Wait()

	// Stop the delayed tasks goroutine
	b.stopDelayedChan <- 1
	// Waiting for the delayed tasks goroutine to have stopped
	b.delayedWG.Wait()

	b.Broker.StopConsuming()

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

	if b.pool != nil {
		b.pool.Close()
	}
}
419189
func (e *endpoint) Close() {
	// Tell dispatch goroutine to stop, then write to the eventfd so that
	// it wakes up in case it's sleeping.
	atomic.StoreUint32(&e.stopRequested, 1)
	syscall.Write(e.rx.eventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})

	// Cleanup the queues inline if the worker hasn't started yet; we also
	// know it won't start from now on because stopRequested is set to 1.
	e.mu.Lock()
	workerPresent := e.workerStarted
	e.mu.Unlock()

	if !workerPresent {
		e.tx.cleanup()
		e.rx.cleanup()
	}
}
1196544
func (reg *WatchDataReg) Close() error {
	reg.adapter.access.Lock()
	defer reg.adapter.access.Unlock()

	for _, sub := range reg.adapter.subscriptions {
		// subscription should have also change channel, otherwise it is not registered
		// for change events
		if sub.ChangeChan != nil && sub.CloseChan != nil {
			// close the channel with all goroutines under subscription
			safeclose.Close(sub.CloseChan)
		}
	}

	delete(reg.adapter.subscriptions, reg.ResyncName)

	return nil
}
1175273
func (s *logScanner) finishLastPointer() *WrappedPointer {
	if s.pointerData.Len() == 0 || !s.currentFileIncluded {
		return nil
	}

	p, err := DecodePointer(s.pointerData)
	s.pointerData.Reset()

	if err == nil {
		return &WrappedPointer{Name: s.currentFilename, Pointer: p}
	} else {
		tracerx.Printf("Unable to parse pointer from log: %v", err)
		return nil
	}
}
1327045
func (*NewGoroutineDispatcher) DispatchTask(task *swf.PollForActivityTaskOutput, handler func(*swf.PollForActivityTaskOutput)) {
	go handler(task)
}
984378
func listenAndDispatch(c *Client, em *clientEventMonitor, r *io.PipeReader, w *io.PipeWriter) {
	listener := make(chan *dockerclient.APIEvents)
	c.dc.AddEventListener(listener)
	for {
		evt := <-listener
		if evt != nil {
			em.dispatch(evt)
		}
	}
}