changes for v0.1.2

* Fixed bug with mixing headers on map buckets rebalancing;
* Added internal option to disable write batching;
* Fixed bug in file requests decoding;
* Downgrade grpc library version;
* All http/2 setting frame options considiration;
* Fixed bug in GOAWAY frame processing;
* Decreased memory allocation for each connection;
* More stream lifecycles capturing;
* Fixed conflict with C# grpc server.

benchmarks/dumb-server/.gitignore

@@ -0,0 +1 @@
+debug

benchmarks/jmeter-java-dsl/docker-compose.yaml

@@ -17,4 +17,4 @@ services:
       resources:
         limits:
           cpus: '2'
-          memory: 3G
+          memory: 2G

cmd/framer/cmd_convert.go

@@ -64,7 +64,7 @@ func (c *ConvertCommand) Run(ctx context.Context) error {
 
 	var r8n reflection.DynamicMessagesStore
 	if c.ReflectionAddr != "" {
-		conn, err := grpc.NewClient(
+		conn, err := grpc.Dial(
 			c.ReflectionAddr,
 			grpc.WithTransportCredentials(insecure.NewCredentials()),
 			grpc.WithUserAgent("framer"),

cmd/framer/cmd_load.go

@@ -132,18 +132,18 @@ func (c *LoadCommand) Run(
 
 	g, ctx := errgroup.WithContext(ctx)
 
-	timeout := consts.DefaultTimeout
-	var reporter types.Reporter = supersimpleReporter.New(timeout)
+	var reporter types.Reporter = supersimpleReporter.New()
 	if c.Phout != "" {
 		f, err := os.Create(c.Phout)
 		if err != nil {
 			return fmt.Errorf("creating phout file(%s): %w", c.Phout, err)
 		}
-		phoutReporter := phoutReporter.New(f, timeout)
+		phoutReporter := phoutReporter.New(f)
 		reporter = multi.NewMutli(phoutReporter, reporter)
 	}
 	g.Go(reporter.Run)
 
+	timeout := consts.DefaultTimeout
 	loaders := make([]*loader.Loader, clients)
 	for i := 0; i < clients; i++ {
 		conn, err := createConn(ctx, timeout, addr)
@@ -154,6 +154,7 @@ func (c *LoadCommand) Run(
 			conn,
 			reporter,
 			timeout,
+			false,
 			log,
 		)
 		if err != nil {

cmd/framer/main_benchmark_test.go

@@ -66,6 +66,7 @@ func BenchmarkE2E(b *testing.B) {
 		conn,
 		reporter,
 		consts.DefaultTimeout,
+		false,
 		zaptest.NewLogger(b),
 	)
 	if err != nil {
@@ -129,6 +130,7 @@ func BenchmarkE2EInMemDatasource(b *testing.B) {
 		conn,
 		reporter,
 		consts.DefaultTimeout,
+		false,
 		zaptest.NewLogger(b),
 	)
 	a.NoError(err)

consts/consts.go

@@ -1,14 +1,17 @@
 package consts
 
-import "time"
+import (
+	"math"
+	"time"
+)
 
 const (
-	ChunksBufferSize    = 2048
 	RecieveBufferSize   = 2048
 	SendBatchTimeout    = time.Millisecond
 	RecieveBatchTimeout = time.Millisecond
 
 	DefaultInitialWindowSize = 65_535
 	DefaultTimeout           = 11 * time.Second
 	DefaultMaxFrameSize      = 16384 // Максимальная длина пейлоада фрейма в grpc. У http2 ограничение больше.
+	DefaultMaxHeaderListSize = math.MaxUint32
 )

datasource/decoder/decoder.go

@@ -0,0 +1,52 @@
+package decoder
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/ozontech/framer/utils/lru"
+)
+
+type Decoder struct {
+	metaUnmarshaler *SafeMultiValDecoder
+
+	tagsLRU    *lru.LRU
+	methodsLRU *lru.LRU
+}
+
+func NewDecoder() *Decoder {
+	return &Decoder{
+		metaUnmarshaler: NewSafeMultiValDecoder(),
+
+		tagsLRU:    lru.New(1024),
+		methodsLRU: lru.New(1024),
+	}
+}
+
+func (decoder *Decoder) Unmarshal(d *Data, b []byte) error {
+	d.Reset()
+
+	tagB, b := nextLine(b)
+	d.Tag = decoder.tagsLRU.GetOrAdd(tagB)
+
+	methodB, b := nextLine(b)
+	d.Method = decoder.methodsLRU.GetOrAdd(methodB)
+
+	metaBytes, b := nextLine(b)
+	var err error
+	d.Metadata, err = decoder.metaUnmarshaler.UnmarshalAppend(d.Metadata, metaBytes)
+	if err != nil {
+		return fmt.Errorf("meta unmarshal error: %w", err)
+	}
+
+	d.Message = b
+	return nil
+}
+
+func nextLine(in []byte) ([]byte, []byte) {
+	index := bytes.IndexByte(in, '\n')
+	if index == -1 {
+		return []byte{}, []byte{}
+	}
+	return in[:index], in[index+1:]
+}

datasource/decoder/jsonkv_safe.go

@@ -0,0 +1,47 @@
+package decoder
+
+import (
+	"github.com/mailru/easyjson/jlexer"
+	"github.com/ozontech/framer/utils/lru"
+)
+
+const (
+	kLRUSize = 1 << 10
+	vLRUSize = 1 << 16
+)
+
+type SafeMultiValDecoder struct {
+	kLRU *lru.LRU
+	vLRU *lru.LRU
+}
+
+func NewSafeMultiValDecoder() *SafeMultiValDecoder {
+	return &SafeMultiValDecoder{
+		lru.New(kLRUSize),
+		lru.New(vLRUSize),
+	}
+}
+
+func (d *SafeMultiValDecoder) UnmarshalAppend(buf []Meta, bytes []byte) ([]Meta, error) {
+	in := jlexer.Lexer{Data: bytes}
+
+	in.Delim('{')
+	for !in.IsDelim('}') {
+		key := d.kLRU.GetOrAdd(in.UnsafeBytes())
+		in.WantColon()
+
+		in.Delim('[')
+		for !in.IsDelim(']') {
+			val := d.kLRU.GetOrAdd(in.UnsafeBytes())
+			buf = append(buf, Meta{Name: key, Value: val})
+			in.WantComma()
+		}
+		in.Delim(']')
+
+		in.WantComma()
+	}
+	in.Delim('}')
+	in.Consumed()
+
+	return buf, in.Error()
+}

datasource/decoder/model.go

@@ -0,0 +1,20 @@
+package decoder
+
+type Meta struct {
+	Name  string
+	Value string
+}
+
+type Data struct {
+	Tag      string
+	Method   string // "/" {service name} "/" {method name}
+	Metadata []Meta
+	Message  []byte
+}
+
+func (d *Data) Reset() {
+	d.Tag = ""
+	d.Method = ""
+	d.Metadata = d.Metadata[:0]
+	d.Message = d.Message[:0]
+}

datasource/file.go

@@ -5,22 +5,27 @@ import (
 	"io"
 	"sync"
 
+	"github.com/ozontech/framer/datasource/decoder"
 	"github.com/ozontech/framer/formats/grpc/ozon/binary"
 	"github.com/ozontech/framer/formats/model"
 	"github.com/ozontech/framer/loader/types"
 	"github.com/ozontech/framer/utils/pool"
 )
 
 type FileDataSource struct {
-	format  *model.InputFormat
+	reader  model.PooledRequestReader
+	decoder *decoder.Decoder
+
 	pool    *pool.SlicePool[*fileRequest]
 	factory *RequestAdapterFactory
 	mu      sync.Mutex
 }
 
 func NewFileDataSource(r io.Reader, factoryOptions ...Option) *FileDataSource {
 	ds := &FileDataSource{
-		binary.NewInput(r),
+		binary.NewInput(r).Reader,
+		decoder.NewDecoder(),
+
 		pool.NewSlicePoolSize[*fileRequest](100),
 		NewRequestAdapterFactory(factoryOptions...),
 		sync.Mutex{},
@@ -32,25 +37,25 @@ func (ds *FileDataSource) Fetch() (types.Req, error) {
 	r, ok := ds.pool.Acquire()
 	if !ok {
 		r = &fileRequest{
-			bytesPool:      ds.format.Reader,
+			bytesPool:      ds.reader,
 			pool:           ds.pool,
 			RequestAdapter: ds.factory.Build(),
 		}
 	}
 
 	var err error
 	ds.mu.Lock()
-	r.bytes, err = ds.format.Reader.ReadNext()
+	r.bytes, err = ds.reader.ReadNext()
 	ds.mu.Unlock()
 	if err != nil {
 		return nil, fmt.Errorf("read next request: %w", err)
 	}
 
-	return r, ds.format.Decoder.Unmarshal(&r.data, r.bytes)
+	return r, ds.decoder.Unmarshal(&r.data, r.bytes)
 }
 
 type fileRequest struct {
-	bytesPool model.PooledRequestReder
+	bytesPool model.PooledRequestReader
 	bytes     []byte
 	pool      *pool.SlicePool[*fileRequest]
 	*RequestAdapter