具有多个goroutine的内存池和缓冲通道

Question

I'm creating a program which create random bson.M documents, and insert them in database. The main goroutine generate the documents, and push them to a buffered channel. In the same time, two goroutines fetch the documents from the channel and insert them in database.

This process take a lot of memory and put too much pressure on garbage colelctor, so I'm trying to implement a memory pool to limit the number of allocations

Here is what I have so far:

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
", 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)
    }
}

But it looks like one List is used 4 times before being regenerated:

> 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]
...

Why isn't the list regenerated each time ? How can I fix this ?

Answer 1

The problem is that you have created a buffered channel with var record = make(chan List, 3). Hence this code:

record <- *list
pool.Put(list)

May return immediately and the entry will be placed back into the pool before it has been consumed. Hence the underlying slice will likely be modified in another loop iteration before your consumer has had a chance to consume it. Although you are sending List as a value object, remember that the []bson.M is a pointer to an allocated array and will still be pointing to the same memory when you send a new List value. Hence why you are seeing the duplicate output.

To fix, modify your channel to send the List pointer make(chan *List, 3) and change your consumer to put the entry back in the pool once finished, e.g:

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

Your producer should then sent the pointer with the pool.Put removed, i.e.

record <- list