Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use Pool to store and reuse GZIPWriters #1329

Open
wants to merge 1 commit into
base: release-branch.v6
Choose a base branch
from
Open

Use Pool to store and reuse GZIPWriters #1329

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ package elastic
import (
"bytes"
"context"
"compress/gzip"
"encoding/json"
"fmt"
"log"
Expand Down Expand Up @@ -102,6 +103,12 @@ var (
noDeprecationLog = func(*http.Request, *http.Response) {}
)

var gzipWriterPool = sync.Pool {
New: func()interface{} {
return gzip.NewWriter(new(bytes.Buffer))
},
}

// ClientOptionFunc is a function that configures a Client.
// It is used in NewClient.
type ClientOptionFunc func(*Client) error
Expand Down Expand Up @@ -1347,7 +1354,18 @@ func (c *Client) PerformRequest(ctx context.Context, opt PerformRequestOptions)

// Set body
if opt.Body != nil {
err = req.SetBody(opt.Body, gzipEnabled)
var gzipWriter *gzip.Writer;
if gzipEnabled {
item := gzipWriterPool.Get()
if item == nil {
c.infof("Got nothing from the pool. Creating a new gzip compressor...")
gzipWriter = gzip.NewWriter(new(bytes.Buffer))
} else {
gzipWriter = item.(*gzip.Writer)
}
defer gzipWriterPool.Put(gzipWriter)
}
err = req.SetBody(opt.Body, gzipEnabled, gzipWriter)
if err != nil {
c.errorf("elastic: couldn't set body %+v for request: %v", opt.Body, err)
return nil, err
Expand Down
14 changes: 7 additions & 7 deletions request.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,16 +37,16 @@ func (r *Request) SetBasicAuth(username, password string) {

// SetBody encodes the body in the request. You may pass a flag to
// compress the request via gzip.
func (r *Request) SetBody(body interface{}, gzipCompress bool) error {
func (r *Request) SetBody(body interface{}, gzipCompress bool, writer *gzip.Writer) error {
switch b := body.(type) {
case string:
if gzipCompress {
return r.setBodyGzip(b)
return r.setBodyGzip(b, writer)
}
return r.setBodyString(b)
default:
if gzipCompress {
return r.setBodyGzip(body)
return r.setBodyGzip(body, writer)
}
return r.setBodyJson(body)
}
Expand All @@ -70,7 +70,7 @@ func (r *Request) setBodyString(body string) error {

// setBodyGzip gzip's the body. It accepts both strings and structs as body.
// The latter will be encoded via json.Marshal.
func (r *Request) setBodyGzip(body interface{}) error {
func (r *Request) setBodyGzip(body interface{}, writer *gzip.Writer) error {
switch b := body.(type) {
case string:
buf := new(bytes.Buffer)
Expand All @@ -90,11 +90,11 @@ func (r *Request) setBodyGzip(body interface{}) error {
return err
}
buf := new(bytes.Buffer)
w := gzip.NewWriter(buf)
if _, err := w.Write(data); err != nil {
writer.Reset(buf)
if _, err := writer.Write(data); err != nil {
return err
}
if err := w.Close(); err != nil {
if err := writer.Close(); err != nil {
return err
}
r.Header.Add("Content-Encoding", "gzip")
Expand Down
18 changes: 11 additions & 7 deletions request_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,11 @@

package elastic

import "testing"
import (
"bytes"
"compress/gzip"
"testing"
)

var testReq *Request // used as a temporary variable to avoid compiler optimizations in tests/benchmarks

Expand All @@ -29,7 +33,7 @@ func BenchmarkRequestSetBodyString(b *testing.B) {
}
for i := 0; i < b.N; i++ {
body := `{"query":{"match_all":{}}}`
err = req.SetBody(body, false)
err = req.SetBody(body, false, nil)
if err != nil {
b.Fatal(err)
}
Expand All @@ -45,7 +49,7 @@ func BenchmarkRequestSetBodyStringGzip(b *testing.B) {
}
for i := 0; i < b.N; i++ {
body := `{"query":{"match_all":{}}}`
err = req.SetBody(body, true)
err = req.SetBody(body, true, gzip.NewWriter(new(bytes.Buffer)))
if err != nil {
b.Fatal(err)
}
Expand All @@ -61,7 +65,7 @@ func BenchmarkRequestSetBodyBytes(b *testing.B) {
}
for i := 0; i < b.N; i++ {
body := []byte(`{"query":{"match_all":{}}}`)
err = req.SetBody(body, false)
err = req.SetBody(body, false, nil)
if err != nil {
b.Fatal(err)
}
Expand All @@ -77,7 +81,7 @@ func BenchmarkRequestSetBodyBytesGzip(b *testing.B) {
}
for i := 0; i < b.N; i++ {
body := []byte(`{"query":{"match_all":{}}}`)
err = req.SetBody(body, true)
err = req.SetBody(body, true, gzip.NewWriter(new(bytes.Buffer)))
if err != nil {
b.Fatal(err)
}
Expand All @@ -97,7 +101,7 @@ func BenchmarkRequestSetBodyMap(b *testing.B) {
"match_all": map[string]interface{}{},
},
}
err = req.SetBody(body, false)
err = req.SetBody(body, false, nil)
if err != nil {
b.Fatal(err)
}
Expand All @@ -117,7 +121,7 @@ func BenchmarkRequestSetBodyMapGzip(b *testing.B) {
"match_all": map[string]interface{}{},
},
}
err = req.SetBody(body, true)
err = req.SetBody(body, true, gzip.NewWriter(new(bytes.Buffer)))
if err != nil {
b.Fatal(err)
}
Expand Down