複数のgoroutinesを持つメモリプールとバッファリングされたチャネル

Question

ランダムなbson.Mドキュメントを作成してデータベースに挿入するプログラムを作成しています。 メインのゴルーチンはドキュメントを生成し、それらをバッファリングされたチャネルにプッシュします。同時に、2つのゴルーチンがチャネルからドキュメントを取り出し、データベースに挿入します。

このプロセスは大量のメモリを取るとガベージcolelctorにあまり圧力をかけるので、私はここで割り当て

の数を制限するために、メモリプールを実装しようとしている私がこれまで持っているものです。

package main 

import (
    "fmt" 
    "math/rand" 
    "sync" 
    "time" 

    "gopkg.in/mgo.v2/bson" 
) 

type List struct { 
    L []bson.M 
} 

func main() { 
    var rndSrc = rand.NewSource(time.Now().UnixNano()) 

    pool := sync.Pool{ 
     New: func() interface{} { 
      l := make([]bson.M, 1000) 
      for i, _ := range l { 
       m := bson.M{} 
       l[i] = m 
      } 
      return &List{L: l} 
     }, 
    } 
    // buffered channel to store generated bson.M docs 
    var record = make(chan List, 3) 
    // start worker to insert docs in database 
    for i := 0; i < 2; i++ { 
     go func() { 
      for r := range record { 
       fmt.Printf("first: %v\n", r.L[0]) 
       // do the insert ect 
      } 
     }() 
    } 
    // feed the channel 
    for i := 0; i < 100; i++ { 
     // get an object from the pool instead of creating a new one 
     list := pool.Get().(*List) 
     // re generate the documents 
     for j, _ := range list.L { 
      list.L[j]["key1"] = rndSrc.Int63() 
     } 
     // push the docs to the channel, and return them to the pool 
     record <- *list 
     pool.Put(list) 
    } 
} 

しかし、1 Listが再生成される前に4回使用されているように見えます：

> go run test.go 
first: map[key1:943279487605002381 key2:4444061964749643436] 
first: map[key1:943279487605002381 key2:4444061964749643436] 
first: map[key1:943279487605002381 key2:4444061964749643436] 
first: map[key1:943279487605002381 key2:4444061964749643436] 
first: map[key1:8767993090152084935 key2:8807650676784718781] 
... 

リストが毎回再生されないのはなぜ？これをどうすれば解決できますか？

Answer 1

問題は、var record = make(chan List, 3)でバッファードチャネルを作成したことです。したがって、このコード：

record <- *list 
pool.Put(list) 

すぐに戻ることができ、エントリは消費される前にプールに戻されます。したがって、消費者がそれを消費する前に、基礎となるスライスが別のループ反復で変更される可能性があります。値オブジェクトとしてListを送信していますが、[]bson.Mは割り当てられた配列へのポインタであり、新しいList値を送信するときには同じメモリを参照していることに注意してください。それでなぜあなたは重複出力を見ているのですか？

for r := range record { 
    fmt.Printf("first: %v\n", r.L[0]) 
    // do the insert etc 
    pool.Put(r) // Even if error occurs 
} 

あなたのプロデューサーは、その後pool.Putでポインタを送らなければならない：例えば、リストポインタmake(chan *List, 3)を送信し、一度終了し、プールに戻ったエントリを置くために、あなたの消費者を変更するには、あなたのチャンネルを修正し、修正するには

削除、すなわち

record <- list