這個例子是從go源碼src/pkg/net/rpc/server_test.go截取出來的
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
<code>func benchmarkendtoendasync(dial func() (*client, error), b *testing.b) {</code>
<code> </code><code>const</code> <code>maxconcurrentcalls = 100</code>
<code> </code><code>b.stoptimer()</code>
<code> </code><code>once.do(startserver)</code>
<code> </code><code>client, err := dial()</code>
<code> </code><code>if</code> <code>err != nil {</code>
<code> </code><code>b.fatal(</code><code>"error dialing:"</code><code>, err)</code>
<code> </code><code>}</code>
<code> </code><code>// asynchronous calls</code>
<code> </code><code>args := &args{7, 8}</code>
<code> </code><code>procs := 4 * runtime.gomaxprocs(-1)</code>
<code> </code><code>send := int32(b.n)</code>
<code> </code><code>recv := int32(b.n)</code>
<code> </code><code>var</code> <code>wg sync.waitgroup</code>
<code> </code><code>wg.add(procs)</code>
<code> </code><code>gate := make(chan bool, maxconcurrentcalls)</code>
<code> </code><code>res := make(chan *call, maxconcurrentcalls)</code>
<code> </code><code>b.starttimer()</code>
<code> </code><code>for</code> <code>p := 0; p < procs; p++ {</code>
<code> </code><code>go func() {</code>
<code> </code><code>for</code> <code>atomic.addint32(&send, -1) >= 0 {</code>
<code> </code><code>gate <- true</code>
<code> </code><code>reply :=</code><code>new</code><code>(reply)</code>
<code> </code><code>client.go(</code><code>"arith.add"</code><code>, args, reply, res)</code>
<code> </code><code>}</code>
<code> </code><code>}()</code>
<code> </code><code>for</code> <code>call := range res {</code>
<code> </code><code>a := call.args.(*args).a</code>
<code> </code><code>b := call.args.(*args).b</code>
<code> </code><code>c := call.reply.(*reply).c</code>
<code> </code><code>if</code> <code>a+b != c {</code>
<code> </code><code>b.fatalf(</code><code>"incorrect reply: add: expected %d got %d"</code><code>, a+b, c)</code>
<code> </code><code>}</code>
<code> </code><code><-gate</code>
<code> </code><code>if</code> <code>atomic.addint32(&recv, -1) == 0 {</code>
<code> </code><code>close(res)</code>
<code> </code><code>wg.done()</code>
<code> </code><code>wg.wait()</code>
<code>}</code>
這個代碼用來對rpc的用戶端go函數進行壓力測試。
這裡有幾個地方值得揣摩下:
先使用startserver(這個函數裡面具體是開啟了一個routine)進行伺服器服務。然後在每個測試用例中啟動server,如果是benchtest的話記得這裡的timer要在啟動伺服器行為之後再開啟。
wg變量是sync.waitgroup類型,add增加計數,done減少計數,wait進行阻塞等待,等計數減為0的時候再停止阻塞。
這裡如果不使用waitgroup進行wait阻塞的話,主routine會先于次routine先結束。會導緻程式提早退出。
是以這裡也給出了一個測試用例中測試異步函數的方法。就是使用waitgroup
看起來gate好像是沒什麼用啊,如果去掉gate呢?有可能會出現“rpc: discarding call reply due to insufficient done chan capacity”
這個gate完全是因為client.go這個函數,rpc包的client.go是異步的調用,雖然是異步調用,這個異步調用的最後一個done參數是一個channel buffer。
當client.go進行完rpc調用後,将信号傳入這個channel buffer。但是這個channel buffer卻是不會阻塞的。
具體看源碼:

這裡select加了個default分支,說明了done是非阻塞的。看注釋,作者認為這個buffer的大小容量應該由調用者來保證。rpc包并不保證容量大小。
方法有個兩個:
這個方法就是gate的使用原因了。隻有gate容量有剩餘的時候才會容許調用client.go
在這個例子中,bench的channel最大隻會是b.n,是以,如果我們配置設定的res的channel buffer大小為b.n也能解決這個問題。
這個方法導緻的效果就是bench的時間變快了,但是mem配置設定增加了。
因為這裡會有多個routine會對send和recive進行操作,這裡就需要保證原子性。
多個并發routine對一個共享變量進行操作有兩種方法,channel和鎖。
這裡當然使用channel也能起到原子操作的效果。sync包的atomic和sync的mutex都是鎖的方式。
是以說這裡其實可以使用channel,mutex,atomic三種方法。
bench test在運作前自身會調用runtime.gomaxprocs進行多核的設定,然後再每個處理器中并行運作測試。
這裡的runtime.gomaxprocs(-1)是擷取你要跑的cpu核數,這個核數是根據bench test的 -test.cpu設定的。具體可以看下src/testing/testing.go parsecpulist。在沒有設定過gomaxprocs和test.cpu的情況下,這裡的runtime.gomaxprocs就預設是1。
你可以使用-test.cpu 1,2,4來設定你的壓力測試用例是有幾個cpu,每個cpu是幾核的。
這裡的procs設定為處理器核數的4倍就是為了測試routine能配置設定遠大于核數的個數,這樣每個核承擔的goroutine能大于1。
上面的for循環就是保證起的routine數是足夠的。