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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
package gui
// channel communication to the plugins
// https://github.com/sourcegraph/conc
// https://www.reddit.com/r/golang/comments/11x1oek/hello_gophers_show_me_your_concurrent_code/
import (
// "regexp"
// "git.wit.org/wit/gui/toolkit"
"sync"
"runtime"
"github.com/sourcegraph/conc"
"github.com/sourcegraph/conc/stream"
"github.com/sourcegraph/conc/panics"
)
// this should never exit
// TODO: clean up all this poorly named code
func makeConc() {
var wg conc.WaitGroup
defer wg.Wait()
startTheThing(&wg)
log(debugError, "panic?")
sleep(2)
log(debugError, "panic? after sleep(5)")
}
func startTheThing(wg *conc.WaitGroup) {
f := func() {
log(debugError, "startTheThing() == about to panic now")
panic("test conc.WaitGroup")
}
wg.Go(func() {
ExampleCatcher(f)
})
}
func ExampleCatcher(f func()) {
var pc panics.Catcher
i := 0
pc.Try(func() { i += 1 })
pc.Try(f)
pc.Try(func() { i += 1 })
recovered := pc.Recovered()
log(debugError, "panic.Recovered():", recovered.Value.(string))
frames := runtime.CallersFrames(recovered.Callers)
for {
frame, more := frames.Next()
log(debugError, "\t", frame.Function)
if !more {
break
}
}
}
func mapStream(
in chan int,
out chan int,
f func(int) int,
) {
tasks := make(chan func())
taskResults := make(chan chan int)
// Worker goroutines
var workerWg sync.WaitGroup
for i := 0; i < 10; i++ {
workerWg.Add(1)
go func() {
defer workerWg.Done()
for task := range tasks {
task()
}
}()
}
// Ordered reader goroutines
var readerWg sync.WaitGroup
readerWg.Add(1)
go func() {
defer readerWg.Done()
for result := range taskResults {
item := <-result
out <- item
}
}()
// Feed the workers with tasks
for elem := range in {
resultCh := make(chan int, 1)
taskResults <- resultCh
tasks <- func() {
resultCh <- f(elem)
}
}
// We've exhausted input.
// Wait for everything to finish
close(tasks)
workerWg.Wait()
close(taskResults)
readerWg.Wait()
}
func mapStream2(
in chan int,
out chan int,
f func(int) int,
) {
s := stream.New().WithMaxGoroutines(10)
for elem := range in {
elem := elem
s.Go(func() stream.Callback {
res := f(elem)
return func() { out <- res }
})
}
s.Wait()
}
|
