Merge pull request #6 from ozontech/dev

changes for v0.1.2

.github/workflows/golangci-ling.yaml

@@ -14,13 +14,13 @@ permissions:
 jobs:
   golangci:
     name: lint
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5
         with:
           go-version: stable
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v6.1.0
         with:
-          version: v1.58
+          version: v1.60.1

benchmarks/dumb-server/.gitignore

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

benchmarks/dumb-server/main.go

@@ -53,7 +53,7 @@ func main() {
 
 	err := g.Wait()
 	if err != nil {
-		fmt.Printf("server exited: " + err.Error())
+		fmt.Println("server exited: " + err.Error())
 		os.Exit(1)
 	}
 }
@@ -543,7 +543,7 @@ type noopFrameProcessor struct{}
 func (p noopFrameProcessor) Process(
 	_ frameheader.FrameHeader,
 	_ []byte,
-	incomplete bool,
+	_ bool,
 ) error {
 	return nil
 }
@@ -601,7 +601,7 @@ type settingsProcessor struct{}
 
 func (p settingsProcessor) Process(
 	header frameheader.FrameHeader,
-	payload []byte,
+	_ []byte,
 	incomplete bool,
 ) error {
 	if incomplete {

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

@@ -22,9 +22,9 @@ var CLI struct {
 
 type VersionFlag string
 
-func (v VersionFlag) Decode(ctx *kong.DecodeContext) error { return nil }
-func (v VersionFlag) IsBool() bool                         { return true }
-func (v VersionFlag) BeforeApply(app *kong.Kong, vars kong.Vars) error {
+func (v VersionFlag) Decode(_ *kong.DecodeContext) error { return nil }
+func (v VersionFlag) IsBool() bool                       { return true }
+func (v VersionFlag) BeforeApply(app *kong.Kong, _ kong.Vars) error {
 	fmt.Println(getVersion())
 	app.Exit(0)
 	return 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 ограничение больше.
+	DefaultMaxFrameSize      = 16384 // DefaultMaxFrameSize - максимальная длина пейлоада фрейма в 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]
+}