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

feat: improve memory management with local buffer caches #845

Merged
merged 1 commit into from
Jul 5, 2024
Merged

feat: improve memory management with local buffer caches #845

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
4 changes: 4 additions & 0 deletions v2/pkg/astjson/astjson.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,10 @@ type JSON struct {
_intSlicePos int
}

func (j *JSON) Size() int {
return len(j.storage)
}

func (j *JSON) Get(nodeRef int, path []string) int {
if len(path) == 0 {
return nodeRef
Expand Down
9 changes: 4 additions & 5 deletions v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"regexp"
"slices"
Expand Down Expand Up @@ -1697,14 +1696,14 @@ func (s *Source) replaceEmptyObject(variables []byte) ([]byte, bool) {
return variables, false
}

func (s *Source) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, writer io.Writer) (err error) {
func (s *Source) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) (err error) {
input = s.compactAndUnNullVariables(input)
return httpclient.DoMultipartForm(s.httpClient, ctx, input, files, writer)
return httpclient.DoMultipartForm(s.httpClient, ctx, input, files, out)
}

func (s *Source) Load(ctx context.Context, input []byte, writer io.Writer) (err error) {
func (s *Source) Load(ctx context.Context, input []byte, out *bytes.Buffer) (err error) {
input = s.compactAndUnNullVariables(input)
return httpclient.Do(s.httpClient, ctx, input, writer)
return httpclient.Do(s.httpClient, ctx, input, out)
}

type GraphQLSubscriptionClient interface {
Expand Down
18 changes: 7 additions & 11 deletions v2/pkg/engine/datasource/httpclient/nethttpclient.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ func releaseBuffer(buf *bytes.Buffer) {
requestBufferPool.Put(buf)
}

func makeHTTPRequest(client *http.Client, ctx context.Context, url, method, headers, queryParams []byte, body io.Reader, enableTrace bool, out io.Writer, contentType string) (err error) {
func makeHTTPRequest(client *http.Client, ctx context.Context, url, method, headers, queryParams []byte, body io.Reader, enableTrace bool, out *bytes.Buffer, contentType string) (err error) {
request, err := http.NewRequestWithContext(ctx, string(method), string(url), body)
if err != nil {
return err
Expand Down Expand Up @@ -204,18 +204,14 @@ func makeHTTPRequest(client *http.Client, ctx context.Context, url, method, head
return err
}

buf := getBuffer()
defer releaseBuffer(buf)

if !enableTrace {
_, err = buf.ReadFrom(respReader)
if err != nil {
return err
}
_, err = buf.WriteTo(out)
_, err = out.ReadFrom(respReader)
return
}

buf := getBuffer()
defer releaseBuffer(buf)

_, err = buf.ReadFrom(respReader)
if err != nil {
return err
Expand Down Expand Up @@ -245,14 +241,14 @@ func makeHTTPRequest(client *http.Client, ctx context.Context, url, method, head
return err
}

func Do(client *http.Client, ctx context.Context, requestInput []byte, out io.Writer) (err error) {
func Do(client *http.Client, ctx context.Context, requestInput []byte, out *bytes.Buffer) (err error) {
url, method, body, headers, queryParams, enableTrace := requestInputParams(requestInput)

return makeHTTPRequest(client, ctx, url, method, headers, queryParams, bytes.NewReader(body), enableTrace, out, ContentTypeJSON)
}

func DoMultipartForm(
client *http.Client, ctx context.Context, requestInput []byte, files []File, out io.Writer,
client *http.Client, ctx context.Context, requestInput []byte, files []File, out *bytes.Buffer,
) (err error) {
if len(files) == 0 {
return errors.New("no files provided")
Expand Down
16 changes: 9 additions & 7 deletions v2/pkg/engine/datasource/introspection_datasource/source.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
package introspection_datasource

import (
"bytes"
"context"
"encoding/json"
"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient"
"io"

"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient"

"github.com/wundergraph/graphql-go-tools/v2/pkg/introspection"
)

Expand All @@ -17,25 +19,25 @@ type Source struct {
introspectionData *introspection.Data
}

func (s *Source) Load(ctx context.Context, input []byte, w io.Writer) (err error) {
func (s *Source) Load(ctx context.Context, input []byte, out *bytes.Buffer) (err error) {
var req introspectionInput
if err := json.Unmarshal(input, &req); err != nil {
return err
}

switch req.RequestType {
case TypeRequestType:
return s.singleType(w, req.TypeName)
return s.singleType(out, req.TypeName)
case TypeEnumValuesRequestType:
return s.enumValuesForType(w, req.OnTypeName, req.IncludeDeprecated)
return s.enumValuesForType(out, req.OnTypeName, req.IncludeDeprecated)
case TypeFieldsRequestType:
return s.fieldsForType(w, req.OnTypeName, req.IncludeDeprecated)
return s.fieldsForType(out, req.OnTypeName, req.IncludeDeprecated)
}

return json.NewEncoder(w).Encode(s.schemaWithoutTypeInfo())
return json.NewEncoder(out).Encode(s.schemaWithoutTypeInfo())
}

func (s *Source) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) (err error) {
func (s *Source) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) (err error) {
panic("not implemented")
}

Expand Down
12 changes: 7 additions & 5 deletions v2/pkg/engine/datasource/pubsub_datasource/pubsub_kafka.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
package pubsub_datasource

import (
"bytes"
"context"
"encoding/json"
"io"

"github.com/buger/jsonparser"
"github.com/cespare/xxhash/v2"
"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient"
"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
"io"
)

type KafkaEventConfiguration struct {
Expand Down Expand Up @@ -65,21 +67,21 @@ type KafkaPublishDataSource struct {
pubSub KafkaPubSub
}

func (s *KafkaPublishDataSource) Load(ctx context.Context, input []byte, w io.Writer) error {
func (s *KafkaPublishDataSource) Load(ctx context.Context, input []byte, out *bytes.Buffer) error {
var publishConfiguration KafkaPublishEventConfiguration
err := json.Unmarshal(input, &publishConfiguration)
if err != nil {
return err
}

if err := s.pubSub.Publish(ctx, publishConfiguration); err != nil {
_, err = io.WriteString(w, `{"success": false}`)
_, err = io.WriteString(out, `{"success": false}`)
return err
}
_, err = io.WriteString(w, `{"success": true}`)
_, err = io.WriteString(out, `{"success": true}`)
return err
}

func (s *KafkaPublishDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) (err error) {
func (s *KafkaPublishDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) (err error) {
panic("not implemented")
}
18 changes: 10 additions & 8 deletions v2/pkg/engine/datasource/pubsub_datasource/pubsub_nats.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
package pubsub_datasource

import (
"bytes"
"context"
"encoding/json"
"io"

"github.com/buger/jsonparser"
"github.com/cespare/xxhash/v2"
"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient"
"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
"io"
)

type NatsStreamConfiguration struct {
Expand Down Expand Up @@ -73,40 +75,40 @@ type NatsPublishDataSource struct {
pubSub NatsPubSub
}

func (s *NatsPublishDataSource) Load(ctx context.Context, input []byte, w io.Writer) error {
func (s *NatsPublishDataSource) Load(ctx context.Context, input []byte, out *bytes.Buffer) error {
var publishConfiguration NatsPublishAndRequestEventConfiguration
err := json.Unmarshal(input, &publishConfiguration)
if err != nil {
return err
}

if err := s.pubSub.Publish(ctx, publishConfiguration); err != nil {
_, err = io.WriteString(w, `{"success": false}`)
_, err = io.WriteString(out, `{"success": false}`)
return err
}

_, err = io.WriteString(w, `{"success": true}`)
_, err = io.WriteString(out, `{"success": true}`)
return err
}

func (s *NatsPublishDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) error {
func (s *NatsPublishDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) error {
panic("not implemented")
}

type NatsRequestDataSource struct {
pubSub NatsPubSub
}

func (s *NatsRequestDataSource) Load(ctx context.Context, input []byte, w io.Writer) error {
func (s *NatsRequestDataSource) Load(ctx context.Context, input []byte, out *bytes.Buffer) error {
var subscriptionConfiguration NatsPublishAndRequestEventConfiguration
err := json.Unmarshal(input, &subscriptionConfiguration)
if err != nil {
return err
}

return s.pubSub.Request(ctx, subscriptionConfiguration, w)
return s.pubSub.Request(ctx, subscriptionConfiguration, out)
}

func (s *NatsRequestDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) error {
func (s *NatsRequestDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) error {
panic("not implemented")
}
8 changes: 4 additions & 4 deletions v2/pkg/engine/datasource/staticdatasource/static_datasource.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
package staticdatasource

import (
"bytes"
"context"
"io"

"github.com/jensneuse/abstractlogger"

Expand Down Expand Up @@ -66,11 +66,11 @@ func (p *Planner[T]) ConfigureSubscription() plan.SubscriptionConfiguration {

type Source struct{}

func (Source) Load(ctx context.Context, input []byte, w io.Writer) (err error) {
_, err = w.Write(input)
func (Source) Load(ctx context.Context, input []byte, out *bytes.Buffer) (err error) {
_, err = out.Write(input)
return
}

func (Source) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) (err error) {
func (Source) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) (err error) {
panic("not implemented")
}
5 changes: 2 additions & 3 deletions v2/pkg/engine/plan/schemausageinfo_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package plan
import (
"bytes"
"context"
"io"
"testing"

"github.com/jensneuse/abstractlogger"
Expand Down Expand Up @@ -487,10 +486,10 @@ type FakeDataSource struct {
source *StatefulSource
}

func (f *FakeDataSource) Load(ctx context.Context, input []byte, w io.Writer) (err error) {
func (f *FakeDataSource) Load(ctx context.Context, input []byte, out *bytes.Buffer) (err error) {
return
}

func (f *FakeDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) (err error) {
func (f *FakeDataSource) LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) (err error) {
return
}
6 changes: 3 additions & 3 deletions v2/pkg/engine/resolve/datasource.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@
package resolve

import (
"bytes"
"context"
"io"

"github.com/cespare/xxhash/v2"
"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/httpclient"
)

type DataSource interface {
Load(ctx context.Context, input []byte, w io.Writer) (err error)
LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, w io.Writer) (err error)
Load(ctx context.Context, input []byte, out *bytes.Buffer) (err error)
LoadWithFiles(ctx context.Context, input []byte, files []httpclient.File, out *bytes.Buffer) (err error)
}

type SubscriptionDataSource interface {
Expand Down
Loading
Loading