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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
package main
import (
"flag"
"fmt"
"log"
"runtime/pprof"
"strings"
"sync"
"time"
)
// A silly Ball type based on the original ping-pong example.
type Ball struct{ hits int }
func main() {
numSecs := flag.Int("s", 1, "Number of seconds game should last")
numPlayers := flag.Int("p", 2, "Number of players")
flag.Parse()
// Believe it or not, one player still makes for a valid game,
// since that player simply checks in with the main goroutine,
// who simply sends the ball back to it.
if *numPlayers < 1 {
log.Fatalf("Invalid number of players: %d", *numPlayers)
}
// Launch the game!
getLeakProfile(func() {
game(*numSecs, *numPlayers)
})
}
// game sends a [Ball] struct to successive players, modeled as
// goroutines, in a ring: imagine the players are the vertices of a
// directed graph in the form of a polygon, and one vertex sends the
// ball, unidirectionally, to its adjacent vertex. One of the vertices
// (the main goroutine) acts as a referee determining whether the game
// should stop (on the basis of a timeout or else some other
// criterion.)
//
// This is a generalization of the ping-pong example.
func game(numSecs, numPlayers int) {
var wg sync.WaitGroup
players := make([]chan Ball, numPlayers+1)
players[0] = make(chan Ball)
for i := range numPlayers {
players[i+1] = player(i+1, &wg, players[i])
}
// Implicitly, there is yet another goroutine here that is
// measuring the criterion for termination: the t channel
// effectively plays the role of a "done" channel here.
t := time.Tick(time.Duration(numSecs) * time.Second)
players[0] <- Ball{}
loop:
for b := range players[len(players)-1] {
fmt.Printf("HITS: %d\n", b.hits)
select {
case players[0] <- b:
case <-t:
break loop
}
}
// This sets off a chain reaction that closes the other
// player-channels (players[1], players[2], etc.)
close(players[0])
// Wait for all of the in-flight player goroutines to
// finish. You need this!
wg.Wait()
// If any of these prints out, we know we did something wrong.
for i, p := range players {
for range p {
log.Fatalf("P%d", i)
}
}
fmt.Println("Done for real!")
}
// player spawns a new player goroutine that reads from the input
// channel. Player returns a channel for listening for the goroutine's
// output. Both channels are technically receive-only, but making them
// bidirectional simplifies making them members of a slice which
// contains a bidirectional channel.
//
// Player goroutines use the provided id parameter for logging when
// they start and stop, which helps sanity-check whether all launched
// goroutines have, in fact, exited.
func player(id int, wg *sync.WaitGroup, input chan Ball) chan Ball {
out := make(chan Ball)
wg.Go(func() {
defer func() {
close(out)
fmt.Printf("(%d) finished\n", id)
}()
fmt.Printf("(%d) started\n", id)
for b := range input {
b.hits++
fmt.Printf("(%d) %d\n", id, b.hits)
// Introduce a mock "blocking" element to the
// computation. This could represent, for
// example, the interval of time through which
// the ball makes contact with the paddle.
time.Sleep(100 * time.Millisecond)
out <- b
}
})
return out
}
// getLeakProfile runs a leaky program snippet, extracts the goroutine
// leak profile, and writes it to stdout.
func getLeakProfile(leakySnippet func()) {
prof := pprof.Lookup("goroutineleak")
defer func() {
time.Sleep(2 * time.Second)
var content strings.Builder
prof.WriteTo(&content, 2)
// Ignore non leaked goroutines
leaks := strings.Split(content.String(), "\n\n")
for _, leak := range leaks {
if strings.Contains(leak, "(leaked)") {
fmt.Println(leak + "\n")
}
}
}()
leakySnippet()
}
|
