File: parallel.go

package info (click to toggle)
golang-github-twotwotwo-sorts 0.0~git20160814.bf5c1f2-2
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, sid
  • size: 172 kB
  • sloc: makefile: 2
file content (81 lines) | stat: -rw-r--r-- 1,705 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
package sorts

import (
	"runtime"
	"sort"
	"sync"
)

// helpers to coordinate parallel sorts

type sortFunc func(sort.Interface, task, func(task))

// MaxProcs controls how many goroutines to start for large sorts. If 0,
// GOMAXPROCS will be used; if 1, all sorts will be serial.
var MaxProcs = 0

// minParallel is the size of the smallest collection we will try to sort in
// parallel.
var minParallel = 10000

// minOffload is the size of the smallest range that can be offloaded to
// another goroutine.
var minOffload = 127

// bufferRatio is how many sorting tasks to queue (buffer) up per
// worker goroutine.
var bufferRatio float32 = 1

// parallelSort calls the sorters with an asyncSort function that will hand
// the task off to another goroutine when possible.
func parallelSort(data sort.Interface, sorter sortFunc, initialTask task) {
	max := runtime.GOMAXPROCS(0)
	if MaxProcs > 0 && MaxProcs < max {
		max = MaxProcs
	}
	l := data.Len()
	if l < minParallel {
		max = 1
	}

	var syncSort func(t task)
	syncSort = func(t task) {
		sorter(data, t, syncSort)
	}
	if max == 1 {
		syncSort(initialTask)
		return
	}

	wg := new(sync.WaitGroup)
	// buffer up one extra task to keep each cpu busy
	sorts := make(chan task, int(float32(max)*bufferRatio))
	var asyncSort func(t task)
	asyncSort = func(t task) {
		if t.end-t.pos < minOffload {
			sorter(data, t, syncSort)
			return
		}
		wg.Add(1)
		select {
		case sorts <- t:
		default:
			sorter(data, t, asyncSort)
			wg.Done()
		}
	}
	doSortWork := func() {
		for task := range sorts {
			sorter(data, task, asyncSort)
			wg.Done()
		}
	}
	for i := 0; i < max; i++ {
		go doSortWork()
	}

	asyncSort(initialTask)

	wg.Wait()
	close(sorts)
}