strings: Join memory leak

[CodeisCold]

What version of Go are you using (go version)?

$ go version
go version go1.17.9 linux/amd64

Does this issue reproduce with the latest release?

yes

What operating system and processor architecture are you using (go env)?

go env Output

$ go env
GO111MODULE=""
GOARCH="amd64"
GOBIN=""
GOCACHE="/root/.cache/go-build"
GOENV="/root/.config/go/env"
GOEXE=""
GOEXPERIMENT=""
GOFLAGS=""
GOHOSTARCH="amd64"
GOHOSTOS="linux"
GOINSECURE=""
GOMODCACHE="/data/bin/codis/pkg/mod"
GONOPROXY=""
GONOSUMDB=""
GOOS="linux"
GOPATH="/data/bin/codis"
GOPRIVATE=""
GOPROXY="https://goproxy.cn,direct"
GOROOT="/usr/local/go"
GOSUMDB="sum.golang.org"
GOTMPDIR=""
GOTOOLDIR="/usr/local/go/pkg/tool/linux_amd64"
GOVCS=""
GOVERSION="go1.17.9"
GCCGO="gccgo"
AR="ar"
CC="gcc"
CXX="g++"
CGO_ENABLED="1"
GOMOD="/dev/null"
CGO_CFLAGS="-g -O2"
CGO_CPPFLAGS=""
CGO_CXXFLAGS="-g -O2"
CGO_FFLAGS="-g -O2"
CGO_LDFLAGS="-g -O2"
PKG_CONFIG="pkg-config"
GOGCCFLAGS="-fPIC -m64 -pthread -fmessage-length=0 -fdebug-prefix-map=/tmp/go-build356646192=/tmp/go-build -gno-record-gcc-switches"

What did you do?

I run elasticsearch bulk updates in goroutine. In goroutine, continually read update params from channel (about 20/s) and generate update string using this(below) code.

// getUpdateStr generate bulk update string of elastic search
func getUpdateStr(chunk []*params, length int) string {
	s := make([]string, length)
	for i, p := range chunk {
		if i >= length {
			break
		}
		s[i] = p.toEsBulkStr()
	}
	return strings.Join(s, "")
}

What did you expect to see?

stable memory

What did you see instead?

increasing memory

[randall77]

What happens to the strings that are returned? Are they being saved somewhere?
The allocations that strings.Join does should only survive a garbage collection if the returned string is retained by something.

[CodeisCold]

strings are used by strings.NewReader:

func (b *BulkConsumer) Consume(batchSize int, interval int, ctx context.Context, wg *sync.WaitGroup) {
	chunk := make([]*params, batchSize)
	i := 0
	timer := time.NewTimer(time.Duration(interval) * time.Second)
	commit := func() {
		defer func() {
			i = 0
		}()

		if i == 0 {
			return
		}
		log.Debug("es update body: " + getUpdateStr(chunk, i))
                 // elasticsearch bulk 
		res, err := b.c.Bulk(strings.NewReader(getUpdateStr(chunk, i)))
		log.Debug(res, err)
		if err != nil || res.StatusCode >= 300 {
			log.Errorf("es bulk fail. res: %s, error: %s", res, err)
		}

		pipe := b.r.Pipeline()
		for j := 0; j < i; j++ {
			pipe.SAdd("member_refresh_tag", chunk[j].memberId)
		}
		_, err = pipe.Exec()
		if err != nil {
			log.Errorf("set member_refresh_tag cache error. error: %s", err)
		}
	}

	for {
		select {
		case <-timer.C:
			commit()
			timer.Reset(time.Duration(interval) * time.Second)
		case p := <-b.ch:
			chunk[i] = &p
			i++
			if i >= batchSize {
				if !timer.Stop() {
					<-timer.C
				}
				commit()
				timer.Reset(time.Duration(interval) * time.Second)
			}
		case <-ctx.Done():
			log.Info("ctx.Done()")
			if !timer.Stop() {
				<-timer.C
			}
			log.Info("bulk_consumer exiting")
			commit()
			log.Info("bulk_consumer exit")
			wg.Done()
			return
		}
	}
}

[randall77]

They end up in b.c.Bulk, whatever that does.

I don't think it is efficient to debug this back and forth on an issue. If you can, please try to build a reproducible example that we can run ourselves.

I suspect something is hanging on to the results. Nothing in the implementation of strings.Join looks suspicious.

[thediveo]

I notice this additional "consumption":

log.Debug("es update body: " + getUpdateStr(chunk, i))

what log do you use?

[CodeisCold]

I notice this additional "consumption":

log.Debug("es update body: " + getUpdateStr(chunk, i))

what log do you use?

logrus

[CodeisCold]

It's a bug of elastic/go-elasticsearch/v7.

Reproducible Code:

package main

import (
	"github.com/elastic/go-elasticsearch/v7"
	"strings"
	"time"
)

func getUpdateStr() string {
	return strings.Join([]string{
		"{\"update\": { \"_id\":    33141, \"_index\": \"member_tag_0\" }}",
		`{"doc": { "channel_uid": 94248 }}`,
	}, "")
}

func main() {
	es, _ := elasticsearch.NewDefaultClient()

	for {
		_, _ = es.Bulk(strings.NewReader(getUpdateStr()))
		time.Sleep(100 * time.Millisecond)
	}
}

[seankhliao]

Closing as not a bug in std

[CodeisCold]

The bug is that I didn't call the Close() method of es Bulk Response.

The flame graph looks weird. Total memory in flame graph is much less than real memory used by project