// Runs HeapSort on a Sortable collection.
// Runs in O(n * lg n) time, but amortizes worse than quicksort
172189
func HeapSort(stuff Sortable) {
	buildHeap(stuff)
	for i := stuff.Len() - 1; i > 0; i-- {
		stuff.Swap(0, i) // put max at end
		shuffleDown(stuff, 0, i)
	}
}
1241839
func (re *runExtractor) Runs() []*platform.Run {
	runs := make([]*platform.Run, 0, len(re.runs))
	for _, r := range re.runs {
		r := r
		runs = append(runs, &r)
	}

	sort.Slice(runs, func(i, j int) bool { return runs[i].ID < runs[j].ID })
	return runs
}
560379
func Run(jobs []Tasker) (err error) {
	// []T does not convert to []Tasker implicitly even is T implements
	// Tasker. We need to iterate on []Tasker making an explicit cast.
	// http://golang.org/doc/faq#convert_slice_of_interface
	prematureEnd := make(chan struct{})
	jobsQueue := make(chan Tasker, workersNumber)
	done := make(chan struct{}, workersNumber)
	var totalDone int
	go populateQueue(jobsQueue, jobs, prematureEnd)
	go parallelizeWorkers(jobsQueue, done)
	// TODO add a case timeout that returns error.
	for {
		select {
		case <-done:
			totalDone++
		case <-prematureEnd:
			err = ErrTasksNotCompleted
		}
		if totalDone == workersNumber {
			// We can assume that jobsQueue is closed and
			// that no goroutine is operating on []Tasker.
			break
		}
	}
	return
}
1004882
func (l sortEntries) qSort() {
	if len(l) > 1 {
		p := partition(l)
		l[:p].qSort()
		l[p+1:].qSort()
	}
}
22622
func (ar *allocRunner) runTasks() {
	for _, task := range ar.tasks {
		go task.Run()
	}

	for _, task := range ar.tasks {
		<-task.WaitCh()
	}
}
1053643
func Sort(g Iterator) *Immutable {
	if g, ok := g.(*Immutable); ok {
		return g
	}
	return build(g, false)
}
39275
func CountingSort(me []int, minVal, maxVal int) (sorted []int) {
	sorted = make([]int, len(me))
	counts := make([]int, maxVal+1-minVal)
	for i := range me {
		counts[me[i]-minVal]++
	}
	for i := 1; i < len(counts); i++ {
		counts[i] += counts[i-1]
	}
	for i := range me {
		sorted[counts[me[i]-minVal]-1] = me[i]
		counts[me[i]-minVal]--
	}
	return
}
852952
func TimSort(data sort.Interface) {
	newArrayTimSorter(data, data.Len()/64).sort(0, data.Len())
}
391525
func HeapInterface(sl interface{}, less interface{}) heap.Interface {
	ret := &sorter{
		sl:   addressableSlice(sl),
		less: reflect.ValueOf(less),
	}
	heap.Init(ret)
	return ret
}
1304095
func (fq *FunctionQueue) run() {
	for {
		select {
		case <-fq.stopCh:
			return
		case f := <-fq.queue:
			retries := 0
			for {
				select {
				case <-fq.stopCh:
					return
				default:
				}
				retries++
				if err := f.f(); err != nil {
					if !f.waitFunc(retries) {
						break
					}
				} else {
					break
				}
			}
		}
	}
}
295226
func (ha *Heap) Update() {
	ha.m.Lock()
	defer ha.m.Unlock()

	runtime.ReadMemStats(&ha.mem)
	ha.g.Update(int64(ha.mem.HeapAlloc))
}
461957
func (a *AllocGarbageCollector) Run() {
	ticker := time.NewTicker(a.config.Interval)
	for {
		select {
		case <-a.triggerCh:
		case <-ticker.C:
		case <-a.shutdownCh:
			ticker.Stop()
			return
		}

		if err := a.keepUsageBelowThreshold(); err != nil {
			a.logger.Error("error garbage collecting allocations", "error", err)
		}
	}
}
400677
func (sm *subscriptionManager) run() error {
	if sm.LoggingEnabled {
		log.Println("SubscriptionManager: Starting run.")
	}
	for {
		// NOTE: we check to see if its time to purge old buffers whenever
		// something happens or a period of inactivity has occurred.
		// An alternative would be to have another goroutine with a
		// select case time.After() but then you'd have concurrency issues
		// with access to the sm.SubEventBuffer and sm.bufferPriorityQueue objs
		// So instead of introducing mutexes we have this uglier manual time check calls
		select {
		case newClient := <-sm.clientSubscriptions:
			sm.handleNewClient(&newClient)
			sm.seeIfTimeToPurgeStaleCategories()
		case disconnected := <-sm.ClientTimeouts:
			sm.handleClientDisconnect(&disconnected)
			sm.seeIfTimeToPurgeStaleCategories()
		case event := <-sm.Events:
			sm.handleNewEvent(&event)
			sm.seeIfTimeToPurgeStaleCategories()
		case <-time.After(time.Duration(5) * time.Second):
			sm.seeIfTimeToPurgeStaleCategories()
		case _ = <-sm.Quit:
			if sm.LoggingEnabled {
				log.Println("SubscriptionManager: received quit signal, stopping.")
			}
			// break out of our infinite loop/select
			return nil
		}
	}
}
84223
func MultithreadedSortComparators(comparators Comparators) Comparators {
	toBeSorted := make(Comparators, len(comparators))
	copy(toBeSorted, comparators)

	var wg sync.WaitGroup

	numCPU := int64(runtime.NumCPU())
	if numCPU%2 == 1 { // single core machine
		numCPU++
	}

	chunks := chunk(toBeSorted, numCPU)
	wg.Add(len(chunks))
	for i := 0; i < len(chunks); i++ {
		go func(i int) {
			sortBucket(chunks[i])
			wg.Done()
		}(i)
	}

	wg.Wait()
	todo := make([]Comparators, len(chunks)/2)
	for {
		todo = todo[:len(chunks)/2]
		wg.Add(len(chunks) / 2)
		for i := 0; i < len(chunks); i += 2 {
			go func(i int) {
				todo[i/2] = SymMerge(chunks[i], chunks[i+1])
				wg.Done()
			}(i)
		}

		wg.Wait()

		chunks = copyChunk(todo)
		if len(chunks) == 1 {
			break
		}
	}

	return chunks[0]
}
1215710
func (s *Collector) Run() {
	type publishersPair struct {
		container *container.Container
		publisher *pubsub.Publisher
	}
	// we cannot determine the capacity here.
	// it will grow enough in first iteration
	var pairs []publishersPair

	for {
		// Put sleep at the start so that it will always be hit,
		// preventing a tight loop if no stats are collected.
		time.Sleep(s.interval)

		// it does not make sense in the first iteration,
		// but saves allocations in further iterations
		pairs = pairs[:0]

		s.m.Lock()
		for container, publisher := range s.publishers {
			// copy pointers here to release the lock ASAP
			pairs = append(pairs, publishersPair{container, publisher})
		}
		s.m.Unlock()
		if len(pairs) == 0 {
			continue
		}

		onlineCPUs, err := s.getNumberOnlineCPUs()
		if err != nil {
			logrus.Errorf("collecting system online cpu count: %v", err)
			continue
		}

		for _, pair := range pairs {
			stats, err := s.supervisor.GetContainerStats(pair.container)

			switch err.(type) {
			case nil:
				// Sample system CPU usage close to container usage to avoid
				// noise in metric calculations.
				systemUsage, err := s.getSystemCPUUsage()
				if err != nil {
					logrus.WithError(err).WithField("container_id", pair.container.ID).Errorf("collecting system cpu usage")
					continue
				}

				// FIXME: move to containerd on Linux (not Windows)
				stats.CPUStats.SystemUsage = systemUsage
				stats.CPUStats.OnlineCPUs = onlineCPUs

				pair.publisher.Publish(*stats)

			case notRunningErr, notFoundErr:
				// publish empty stats containing only name and ID if not running or not found
				pair.publisher.Publish(types.StatsJSON{
					Name: pair.container.Name,
					ID:   pair.container.ID,
				})

			default:
				logrus.Errorf("collecting stats for %s: %v", pair.container.ID, err)
				pair.publisher.Publish(types.StatsJSON{
					Name: pair.container.Name,
					ID:   pair.container.ID,
				})
			}
		}
	}
}
1220830
func (s *Scheduler) RunAll() {
	for i := 0; i < s.size; i++ {
		s.jobs[i].run()
	}
}
1107030
func (eq *eventQueue) run() {
	for {
		block := eq.next()

		if block == nil {
			return // nil block means event queue is closed.
		}

		if err := eq.sink.Write(block...); err != nil {
			logrus.Warnf("eventqueue: error writing events to %v, these events will be lost: %v", eq.sink, err)
		}

		for _, listener := range eq.listeners {
			listener.egress(block...)
		}
	}
}