Repository: cloudevents/sdk-go
Pull request #284 (2020-01-20 13:47:27)

Fix: defeat http.Client memory leak
u should check t.Client is nil or not again in mutex lock


Patch:
diff --git a/pkg/cloudevents/transport/http/transport.go b/pkg/cloudevents/transport/http/transport.go
index 1920304..f68abb1 100644
--- a/pkg/cloudevents/transport/http/transport.go
+++ b/pkg/cloudevents/transport/http/transport.go
@@ -170,7 +170,9 @@ func (t *Transport) Send(ctx context.Context, event cloudevents.Event) (context.
 func (t *Transport) obsSend(ctx context.Context, event cloudevents.Event) (context.Context, *cloudevents.Event, error) {
 	if t.Client == nil {
 		t.crMu.Lock()
-		t.Client = &http.Client{}
+		if t.Client == nil {
+			t.Client = &http.Client{}
+		}
 		t.crMu.Unlock()
 	}
 
diff --git a/pkg/cloudevents/transport/pubsub/codec_v1.go b/pkg/cloudevents/transport/pubsub/codec_v1.go
index 85396de..2d09101 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v1.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v1.go
@@ -2,7 +2,6 @@ package pubsub
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"strings"
 
@@ -118,27 +117,13 @@ func (v CodecV1) toAttributes(e cloudevents.Event) (map[string]string, error) {
 	}
 
 	for k, v := range e.Extensions() {
-		if mapVal, ok := v.(map[string]interface{}); ok {
-			for subkey, subval := range mapVal {
-				encoded, err := json.Marshal(subval)
-				if err != nil {
-					return nil, err
-				}
-				a[prefix+k+"-"+subkey] = string(encoded)
-			}
-			continue
-		}
-		if s, ok := v.(string); ok {
-			a[prefix+k] = s
-			continue
-		}
-		encoded, err := json.Marshal(v)
+		k = strings.ToLower(k)
+		cstr, err := types.Format(v)
 		if err != nil {
-			return nil, err
+			return a, err
 		}
-		a[prefix+k] = string(encoded)
+		a[prefix+k] = cstr
 	}
-
 	return a, nil
 }
 
@@ -244,32 +229,7 @@ func (v CodecV1) fromAttributes(a map[string]string, event *cloudevents.Event) e
 	for k, v := range a {
 		if len(k) > len(prefix) && strings.EqualFold(k[:len(prefix)], prefix) {
 			ak := strings.ToLower(k[len(prefix):])
-			if i := strings.Index(ak, "-"); i > 0 {
-				// attrib-key
-				attrib := ak[:i]
-				key := ak[(i + 1):]
-				if xv, ok := extensions[attrib]; ok {
-					if m, ok := xv.(map[string]interface{}); ok {
-						m[key] = v
-						continue
-					}
-					// TODO: revisit how we want to bubble errors up.
-					return fmt.Errorf("failed to process map type extension")
-				} else {
-					m := make(map[string]interface{})
-					m[key] = v
-					extensions[attrib] = m
-				}
-			} else {
-				// key
-				var tmp interface{}
-				if err := json.Unmarshal([]byte(v), &tmp); err == nil {
-					extensions[ak] = tmp
-				} else {
-					// If we can't unmarshal the data, treat it as a string.
-					extensions[ak] = v
-				}
-			}
+			extensions[ak] = v
 		}
 	}
 	event.Context = ec


Test patch:
diff --git a/pkg/cloudevents/transport/pubsub/codec_test.go b/pkg/cloudevents/transport/pubsub/codec_test.go
index ca58a38..12af951 100644
--- a/pkg/cloudevents/transport/pubsub/codec_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_test.go
@@ -14,6 +14,10 @@ import (
 	"github.com/google/go-cmp/cmp"
 )
 
+const (
+	prefix = "ce-"
+)
+
 func strptr(s string) *string {
 	return &s
 }
diff --git a/pkg/cloudevents/transport/pubsub/codec_v1_test.go b/pkg/cloudevents/transport/pubsub/codec_v1_test.go
index 0b4dfb7..255b620 100644
--- a/pkg/cloudevents/transport/pubsub/codec_v1_test.go
+++ b/pkg/cloudevents/transport/pubsub/codec_v1_test.go
@@ -62,7 +62,8 @@ func TestCodecV1_Encode(t *testing.T) {
 					DataContentType: cloudevents.StringOfApplicationJSON(),
 					Source:          *source,
 					Extensions: map[string]interface{}{
-						"test": "extended",
+						"test":       "extended",
+						"generation": "1579743478182200",
 					},
 				}.AsV1(),
 				Data: map[string]interface{}{
@@ -84,6 +85,7 @@ func TestCodecV1_Encode(t *testing.T) {
 						"time":       now,
 						"type":       "com.example.test",
 						"test":       "extended",
+						"generation": "1579743478182200",
 						"dataschema": "http://example.com/schema",
 						"source":     "http://example.com/source",
 					}
@@ -91,6 +93,46 @@ func TestCodecV1_Encode(t *testing.T) {
 				}(),
 			},
 		},
+		"full v1.0 binary": {
+			codec: pubsub.CodecV1{
+				DefaultEncoding: pubsub.BinaryV1,
+			},
+			event: cloudevents.Event{
+				Context: cloudevents.EventContextV1{
+					ID:         "ABC-123",
+					Time:       &now,
+					Type:       "com.example.test",
+					DataSchema: schema,
+					Source:     *source,
+					Extensions: map[string]interface{}{
+						"test":       "extended",
+						"generation": "1579743478182200",
+					},
+				}.AsV1(),
+				Data: map[string]interface{}{
+					"hello": "world",
+				},
+			},
+			want: &pubsub.Message{
+				Attributes: map[string]string{
+					prefix + "specversion":     "1.0",
+					prefix + "id":              "ABC-123",
+					prefix + "datacontenttype": cloudevents.ApplicationJSON,
+					prefix + "time":            now.String(),
+					prefix + "type":            "com.example.test",
+					prefix + "dataschema":      "http://example.com/schema",
+					prefix + "source":          "http://example.com/source",
+					prefix + "test":            "extended",
+					prefix + "generation":      "1579743478182200",
+				},
+				Data: func() []byte {
+					data := map[string]interface{}{
+						"hello": "world",
+					}
+					return toBytes(data)
+				}(),
+			},
+		},
 		"simple v1.0 structured": {
 			codec: pubsub.CodecV1{DefaultEncoding: pubsub.StructuredV1},
 			event: cloudevents.Event{
@@ -238,6 +280,7 @@ func TestCodecV1_Decode(t *testing.T) {
 					"time":       now,
 					"type":       "com.example.test",
 					"test":       "extended",
+					"generation": "1579743478182200",
 					"dataschema": "http://example.com/schema",
 					"source":     "http://example.com/source",
 				}),
@@ -252,7 +295,44 @@ func TestCodecV1_Decode(t *testing.T) {
 					DataContentType: cloudevents.StringOfApplicationJSON(),
 					Source:          *source,
 					Extensions: map[string]interface{}{
-						"test": "extended",
+						"test":       "extended",
+						"generation": "1579743478182200",
+					},
+				},
+				Data: toBytes(map[string]interface{}{
+					"hello": "world",
+				}),
+				DataEncoded: true,
+			},
+		},
+		"full v1.0 binary": {
+			codec: pubsub.CodecV1{},
+			msg: &pubsub.Message{
+				Attributes: map[string]string{
+					prefix + "specversion": "1.0",
+					prefix + "id":          "ABC-123",
+					prefix + "time":        now.String(),
+					prefix + "type":        "com.example.test",
+					prefix + "dataschema":  "http://example.com/schema",
+					prefix + "source":      "http://example.com/source",
+					prefix + "test":        "extended",
+					prefix + "generation":  "1579743478182200",
+				},
+				Data: toBytes(map[string]interface{}{
+					"hello": "world",
+				}),
+			},
+			want: &cloudevents.Event{
+				Context: &cloudevents.EventContextV1{
+					SpecVersion: cloudevents.CloudEventsVersionV1,
+					ID:          "ABC-123",
+					Time:        &now,
+					Type:        "com.example.test",
+					DataSchema:  schema,
+					Source:      *source,
+					Extensions: map[string]interface{}{
+						"test":       "extended",
+						"generation": "1579743478182200",
 					},
 				},
 				Data: toBytes(map[string]interface{}{
diff --git a/test/benchmark/benchmark_case.go b/test/benchmark/benchmark_case.go
new file mode 100644
index 0000000..fdfade6
--- /dev/null
+++ b/test/benchmark/benchmark_case.go
@@ -0,0 +1,28 @@
+package benchmark
+
+type BenchmarkCase struct {
+	PayloadSize   int
+	Parallelism   int
+	OutputSenders int
+}
+
+func GenerateAllBenchmarkCases(
+	payloadMin int,
+	payloadMax int,
+	parallelismMin int,
+	parallelismMax int,
+	outputSendersMin int,
+	outputSendersMax int,
+) []BenchmarkCase {
+	var cases []BenchmarkCase
+
+	for payload := payloadMin; payload <= payloadMax; payload *= 2 {
+		for parallelism := parallelismMin; parallelism <= parallelismMax; parallelism += 1 {
+			for outputSenders := outputSendersMin; outputSenders <= outputSendersMax; outputSenders += 1 {
+				cases = append(cases, BenchmarkCase{payload, parallelism, outputSenders})
+			}
+		}
+	}
+
+	return cases
+}
diff --git a/test/benchmark/benchmark_result.go b/test/benchmark/benchmark_result.go
new file mode 100644
index 0000000..1ee3242
--- /dev/null
+++ b/test/benchmark/benchmark_result.go
@@ -0,0 +1,34 @@
+package benchmark
+
+import (
+	"encoding/csv"
+	"strconv"
+	"testing"
+)
+
+type BenchmarkResult struct {
+	BenchmarkCase
+	testing.BenchmarkResult
+}
+
+func (br *BenchmarkResult) record() []string {
+	return []string{
+		strconv.Itoa(br.Parallelism),
+		strconv.Itoa(br.PayloadSize),
+		strconv.Itoa(br.OutputSenders),
+		strconv.FormatInt(br.NsPerOp(), 10),
+		strconv.FormatInt(br.AllocedBytesPerOp(), 10),
+	}
+}
+
+type BenchmarkResults []BenchmarkResult
+
+func (br BenchmarkResults) WriteToCsv(writer *csv.Writer) error {
+	for _, i2 := range br {
+		err := writer.Write(i2.record())
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/test/benchmark/http/3d_plot_allocs.gnuplot b/test/benchmark/http/3d_plot_allocs.gnuplot
new file mode 100644
index 0000000..4972abb
--- /dev/null
+++ b/test/benchmark/http/3d_plot_allocs.gnuplot
@@ -0,0 +1,19 @@
+set datafile separator comma
+set datafile missing NaN
+set xlabel "Parallelism"
+set ylabel "Number of output senders"
+set zlabel "Memory Allocated/Ops"
+payload_size_kb=ARG1
+payload_size=payload_size_kb*1024
+
+print "Plotting with payload size ".payload_size.""
+
+splot "baseline-binary.csv" using 1:3:($2==payload_size?$5:1/0) title "Baseline Binary ".payload_size_kb."kb" with linespoint, \
+     "baseline-structured.csv" using 1:3:($2==payload_size?$5:1/0) title "Baseline Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-structured.csv" using 1:3:($2==payload_size?$5:1/0) title "Binding Structured to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-binary.csv" using 1:3:($2==payload_size?$5:1/0) title "Binding Structured to Binary ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-structured.csv" using 1:3:($2==payload_size?$5:1/0) title "Binding Binary to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-binary.csv" using 1:3:($2==payload_size?$5:1/0) title "Binding Binary to Binary ".payload_size_kb."kb" with linespoint, \
+     "client-binary.csv" using 1:3:($2==payload_size?$5:1/0) title "Client Binary ".payload_size_kb."kb" with linespoint, \
+     "client-structured.csv" using 1:3:($2==payload_size?$5:1/0) title "Client Structured ".payload_size_kb."kb" with linespoint
+pause -1
diff --git a/test/benchmark/http/3d_plot_ns.gnuplot b/test/benchmark/http/3d_plot_ns.gnuplot
new file mode 100644
index 0000000..8b5fd32
--- /dev/null
+++ b/test/benchmark/http/3d_plot_ns.gnuplot
@@ -0,0 +1,19 @@
+set datafile separator comma
+set datafile missing NaN
+set xlabel "Parallelism"
+set ylabel "Number of output senders"
+set zlabel "Nanoseconds/Ops"
+payload_size_kb=ARG1
+payload_size=payload_size_kb*1024
+
+print "Plotting with payload size ".payload_size.""
+
+splot "baseline-binary.csv" using 1:3:($2==payload_size?$4:1/0) title "Baseline Binary ".payload_size_kb."kb" with linespoint, \
+     "baseline-structured.csv" using 1:3:($2==payload_size?$4:1/0) title "Baseline Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-structured.csv" using 1:3:($2==payload_size?$4:1/0) title "Binding Structured to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-binary.csv" using 1:3:($2==payload_size?$4:1/0) title "Binding Structured to Binary ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-structured.csv" using 1:3:($2==payload_size?$4:1/0) title "Binding Binary to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-binary.csv" using 1:3:($2==payload_size?$4:1/0) title "Binding Binary to Binary ".payload_size_kb."kb" with linespoint, \
+     "client-binary.csv" using 1:3:($2==payload_size?$4:1/0) title "Client Binary ".payload_size_kb."kb" with linespoint, \
+     "client-structured.csv" using 1:3:($2==payload_size?$4:1/0) title "Client Structured ".payload_size_kb."kb" with linespoint
+pause -1
diff --git a/test/benchmark/http/http_mock.go b/test/benchmark/http/http_mock.go
new file mode 100644
index 0000000..1101f74
--- /dev/null
+++ b/test/benchmark/http/http_mock.go
@@ -0,0 +1,94 @@
+package main
+
+import (
+	"bytes"
+	"io"
+	"io/ioutil"
+	nethttp "net/http"
+	"net/http/httptest"
+	"net/url"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+	"github.com/cloudevents/sdk-go/pkg/binding"
+	"github.com/cloudevents/sdk-go/pkg/bindings/http"
+	cehttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
+)
+
+type RoundTripFunc func(req *nethttp.Request) *nethttp.Response
+
+func (f RoundTripFunc) RoundTrip(req *nethttp.Request) (*nethttp.Response, error) {
+	return f(req), nil
+}
+
+func NewTestClient(fn RoundTripFunc) *nethttp.Client {
+	return &nethttp.Client{
+		Transport: RoundTripFunc(fn),
+	}
+}
+
+func MockedSender(options ...http.SenderOptionFunc) binding.Sender {
+	u, _ := url.Parse("http://localhost")
+	return http.NewSender(NewTestClient(func(req *nethttp.Request) *nethttp.Response {
+		return &nethttp.Response{
+			StatusCode: 202,
+			Header:     make(nethttp.Header),
+		}
+	}), u, options...)
+}
+
+func MockedClient() (cloudevents.Client, *cehttp.Transport) {
+	t, err := cehttp.New(cehttp.WithTarget("http://localhost"))
+
+	if err != nil {
+		panic(err)
+	}
+
+	t.Client = NewTestClient(func(req *nethttp.Request) *nethttp.Response {
+		return &nethttp.Response{
+			StatusCode: 202,
+			Header:     make(nethttp.Header),
+			Body:       ioutil.NopCloser(bytes.NewReader([]byte{})),
+		}
+	})
+
+	client, err := cloudevents.NewClient(t)
+
+	if err != nil {
+		panic(err)
+	}
+
+	return client, t
+}
+
+func MockedBinaryRequest(body []byte) *nethttp.Request {
+	r := httptest.NewRequest("POST", "http://localhost:8080", bytes.NewBuffer(body))
+	r.Header.Add("Ce-id", "0")
+	r.Header.Add("Ce-subject", "sub")
+	r.Header.Add("Ce-specversion", "1.0")
+	r.Header.Add("Ce-type", "t")
+	r.Header.Add("Ce-source", "http://localhost")
+	r.Header.Add("Content-type", "text/plain")
+	return r
+}
+
+var (
+	eventBegin = []byte("{" +
+		"\"id\":\"0\"," +
+		"\"subject\":\"sub\"," +
+		"\"specversion\":\"1.0\"," +
+		"\"type\":\"t\"," +
+		"\"source\":\"http://localhost\"," +
+		"\"datacontenttype\":\"text/plain\"," +
+		"\"data\": \"")
+	eventEnd = []byte("\"}")
+)
+
+func MockedStructuredRequest(body []byte) *nethttp.Request {
+	r := httptest.NewRequest(
+		"POST",
+		"http://localhost:8080",
+		io.MultiReader(bytes.NewReader(eventBegin), bytes.NewBuffer(body), bytes.NewReader(eventEnd)),
+	)
+	r.Header.Add("Content-type", cloudevents.ApplicationCloudEventsJSON)
+	return r
+}
diff --git a/test/benchmark/http/main.go b/test/benchmark/http/main.go
index 44c1e8e..1c4a2dd 100644
--- a/test/benchmark/http/main.go
+++ b/test/benchmark/http/main.go
@@ -1,198 +1,194 @@
 package main
 
 import (
-	"bytes"
 	"context"
 	"encoding/csv"
 	"flag"
+	"fmt"
 	"io"
-	"io/ioutil"
+	"math/rand"
 	nethttp "net/http"
 	"net/http/httptest"
-	"net/url"
 	"os"
 	"runtime"
 	"runtime/pprof"
-	"strconv"
+	"sync"
 	"testing"
+	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go"
 	"github.com/cloudevents/sdk-go/pkg/binding"
 	"github.com/cloudevents/sdk-go/pkg/bindings/http"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
-	cehttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
+	"github.com/cloudevents/sdk-go/test/benchmark"
 )
 
-type RoundTripFunc func(req *nethttp.Request) *nethttp.Response
+var letters = []byte("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
 
-func (f RoundTripFunc) RoundTrip(req *nethttp.Request) (*nethttp.Response, error) {
-	return f(req), nil
-}
-
-func NewTestClient(fn RoundTripFunc) *nethttp.Client {
-	return &nethttp.Client{
-		Transport: RoundTripFunc(fn),
+func fillRandom(buf []byte, r *rand.Rand) {
+	for i := 0; i < cap(buf); i++ {
+		buf[i] = letters[r.Intn(len(letters))]
 	}
 }
 
-func generateRandomValue(kb int, value byte) []byte {
-	length := 1024 * kb
-	b := make([]byte, length)
-	for i := 0; i < length; i++ {
-		b[i] = value
-	}
-	return b
-}
+// Avoid DCE
+var W *httptest.ResponseRecorder
+var R *nethttp.Request
 
-func MockedSender() binding.Sender {
-	u, _ := url.Parse("http://localhost")
-	return http.NewSender(NewTestClient(func(req *nethttp.Request) *nethttp.Response {
-		return &nethttp.Response{
-			StatusCode: 202,
-			Header:     make(nethttp.Header),
+func benchmarkBaseline(cases []benchmark.BenchmarkCase, requestFactory func([]byte) *nethttp.Request) benchmark.BenchmarkResults {
+	var results benchmark.BenchmarkResults
+	r := rand.New(rand.NewSource(time.Now().Unix()))
+
+	for _, c := range cases {
+		if c.OutputSenders > 1 {
+			// It doesn't make sense for this test
+			continue
 		}
-	}), u)
-}
+		fmt.Printf("%+v\n", c)
 
-func MockedClient() (cloudevents.Client, *cehttp.Transport) {
-	t, err := cehttp.New(cehttp.WithTarget("http://localhost"))
+		buffer := make([]byte, c.PayloadSize)
+		fillRandom(buffer, r)
 
-	if err != nil {
-		panic(err)
+		result := testing.Benchmark(func(b *testing.B) {
+			b.SetParallelism(c.Parallelism)
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					W = httptest.NewRecorder()
+					R = requestFactory(buffer)
+				}
+			})
+		})
+		results = append(results, benchmark.BenchmarkResult{BenchmarkCase: c, BenchmarkResult: result})
 	}
 
-	t.Client = NewTestClient(func(req *nethttp.Request) *nethttp.Response {
-		return &nethttp.Response{
-			StatusCode: 202,
-			Header:     make(nethttp.Header),
-			Body:       ioutil.NopCloser(bytes.NewReader([]byte{})),
-		}
-	})
-
-	client, err := cloudevents.NewClient(t)
+	return results
+}
 
-	if err != nil {
-		panic(err)
+func pipeLoop(r *http.Receiver, endCtx context.Context, outputSenders int, opts ...http.SenderOptionFunc) {
+	s := MockedSender(opts...)
+	var err error
+	var m binding.Message
+	messageCtx := context.Background()
+	for err != io.EOF {
+		select {
+		case _, ok := <-endCtx.Done():
+			if !ok {
+				return
+			}
+		default:
+			m, err = r.Receive(messageCtx)
+			if err != nil {
+				continue
+			}
+			//TODO multisend when implemented!
+			//Be aware that multisend implementation MUST finish the message AFTER all sends are completed,
+			//otherwise we cheat the results, because the time is calculated from the message received to Message.finish() call
+			_ = s.Send(messageCtx, m)
+		}
 	}
+}
 
-	t.SetReceiver(transport.ReceiveFunc(func(ctx context.Context, e cloudevents.Event, er *cloudevents.EventResponse) error {
-		_, _, _ = client.Send(ctx, e)
-		er.RespondWith(202, nil)
-		return nil
-	}))
+func benchmarkReceiverSender(cases []benchmark.BenchmarkCase, requestFactory func([]byte) *nethttp.Request, opts ...http.SenderOptionFunc) benchmark.BenchmarkResults {
+	var results benchmark.BenchmarkResults
+	random := rand.New(rand.NewSource(time.Now().Unix()))
 
-	return client, t
-}
+	for _, c := range cases {
+		fmt.Printf("%+v\n", c)
 
-func MockedRequest(body []byte) *nethttp.Request {
-	r := httptest.NewRequest("POST", "http://localhost:8080", bytes.NewBuffer(body))
-	r.Header.Add("Ce-id", "0")
-	r.Header.Add("Ce-subject", "sub")
-	r.Header.Add("Ce-specversion", "1.0")
-	r.Header.Add("Ce-type", "t")
-	r.Header.Add("Ce-source", "http://localhost")
-	r.Header.Add("Content-type", "text/plain")
-	return r
-}
+		ctx, cancel := context.WithCancel(context.TODO())
+		receiver := http.NewReceiver()
 
-// Avoid DCE
-var W *httptest.ResponseRecorder
-var R *nethttp.Request
+		// Spawn dispatchers
+		for i := 0; i < c.Parallelism; i++ {
+			go pipeLoop(receiver, ctx, c.OutputSenders, opts...)
+		}
 
-type BenchResult struct {
-	parallelism   int
-	payloadSizeKb int
-	testing.BenchmarkResult
-}
+		buffer := make([]byte, c.PayloadSize)
+		fillRandom(buffer, random)
+		runtime.GC()
 
-func runBench(do func(body []byte)) []BenchResult {
-	results := make([]BenchResult, 0)
-	for p := 1; p <= runtime.NumCPU(); p++ {
-		for k := 1; k <= 32; k *= 2 {
-			body := generateRandomValue(k, byte('a'))
-			r := testing.Benchmark(func(b *testing.B) {
-				b.SetParallelism(p)
-				b.RunParallel(func(pb *testing.PB) {
-					for pb.Next() {
-						do(body)
-					}
-				})
+		result := testing.Benchmark(func(b *testing.B) {
+			b.SetParallelism(c.Parallelism)
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					w := httptest.NewRecorder()
+					receiver.ServeHTTP(w, requestFactory(buffer))
+				}
 			})
-			results = append(results, BenchResult{p, k, r})
-		}
+		})
+		results = append(results, benchmark.BenchmarkResult{BenchmarkCase: c, BenchmarkResult: result})
+
+		cancel()
+		runtime.GC()
 	}
+
 	return results
 }
 
-func benchmarkBaseline() []BenchResult {
-	return runBench(func(body []byte) {
-		W = httptest.NewRecorder()
-		R = MockedRequest(body)
+func dispatchReceiver(clients []cloudevents.Client, outputSenders int) transport.Receiver {
+	return transport.ReceiveFunc(func(ctx context.Context, e cloudevents.Event, er *cloudevents.EventResponse) error {
+		var wg sync.WaitGroup
+		for i := 0; i < outputSenders; i++ {
+			wg.Add(1)
+			go func(client cloudevents.Client) {
+				_, _, _ = client.Send(ctx, e)
+				wg.Done()
+			}(clients[i])
+		}
+		wg.Wait()
+		er.RespondWith(200, nil)
+		return nil
 	})
 }
 
-func benchmarkReceiverSender() []BenchResult {
-	r := http.NewReceiver()
+func benchmarkClient(cases []benchmark.BenchmarkCase, requestFactory func([]byte) *nethttp.Request) benchmark.BenchmarkResults {
+	var results benchmark.BenchmarkResults
+	random := rand.New(rand.NewSource(time.Now().Unix()))
 
-	results := make([]BenchResult, 0)
-	for p := 1; p <= runtime.NumCPU(); p++ {
-		ctx, cancel := context.WithCancel(context.TODO())
+	for _, c := range cases {
+		fmt.Printf("%+v\n", c)
 
-		// Spawn dispatchers
-		for i := 0; i < p; i++ {
-			go func(r *http.Receiver) {
-				s := MockedSender()
-				var err error
-				var m binding.Message
-				messageCtx := context.Background()
-				for err != io.EOF {
-					select {
-					case _, ok := <-ctx.Done():
-						if !ok {
-							return
-						}
-					default:
-						m, err = r.Receive(messageCtx)
-						if err != nil {
-							continue
-						}
-						_ = s.Send(messageCtx, m)
-					}
-				}
-			}(r)
+		_, mockedReceiverTransport := MockedClient()
+
+		senderClients := make([]cloudevents.Client, c.OutputSenders)
+		for i := 0; i < c.OutputSenders; i++ {
+			senderClients[i], _ = MockedClient()
 		}
 
-		for k := 1; k <= 32; k *= 2 {
-			body := generateRandomValue(k, byte('a'))
-			r := testing.Benchmark(func(b *testing.B) {
-				b.SetParallelism(p)
-				b.RunParallel(func(pb *testing.PB) {
-					for pb.Next() {
-						w := httptest.NewRecorder()
-						r.ServeHTTP(w, MockedRequest(body))
-					}
-				})
+		mockedReceiverTransport.SetReceiver(dispatchReceiver(senderClients, c.OutputSenders))
+
+		buffer := make([]byte, c.PayloadSize)
+		fillRandom(buffer, random)
+		runtime.GC()
+
+		result := testing.Benchmark(func(b *testing.B) {
+			b.SetParallelism(c.Parallelism)
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					w := httptest.NewRecorder()
+					mockedReceiverTransport.ServeHTTP(w, requestFactory(buffer))
+				}
 			})
-			results = append(results, BenchResult{p, k, r})
-		}
+		})
+		results = append(results, benchmark.BenchmarkResult{BenchmarkCase: c, BenchmarkResult: result})
 
-		cancel()
+		runtime.GC()
 	}
-	return results
-}
-
-func benchmarkClient() []BenchResult {
-	_, mockedTransport := MockedClient()
 
-	return runBench(func(body []byte) {
-		w := httptest.NewRecorder()
-		mockedTransport.ServeHTTP(w, MockedRequest(body))
-	})
+	return results
 }
 
 var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to `file`")
 var memprofile = flag.String("memprofile", "", "write memory profile to `file`")
-var bench = flag.String("bench", "baseline", "[baseline, receiver-sender, client]")
+var bench = flag.String(
+	"bench",
+	"baseline-binary",
+	"[baseline-structured, baseline-binary, binding-structured-to-structured, binding-structured-to-binary, binding-binary-to-structured, binding-binary-to-binary, client-binary, client-structured]",
+)
+var out = flag.String("out", "out.csv", "Output file")
+var maxPayloadKb = flag.Int("max-payload", 32, "Max payload size in kb")
+var maxParallelism = flag.Int("max-parallelism", runtime.NumCPU()*2, "Max parallelism")
+var maxOutputSenders = flag.Int("max-output-senders", 1, "Max output senders")
 
 func main() {
 	flag.Parse()
@@ -204,17 +200,43 @@ func main() {
 		defer pprof.StopCPUProfile()
 	}
 
-	var results []BenchResult
+	benchmarkCases := benchmark.GenerateAllBenchmarkCases(
+		1024,
+		1024*(*maxPayloadKb),
+		1,
+		*maxParallelism,
+		1,
+		*maxOutputSenders, //TODO to be increased when receiver-sender will support multi senders
+	)
+
+	var results benchmark.BenchmarkResults
+
+	fmt.Printf("--- Starting benchmark %s ---\n", *bench)
 
 	switch *bench {
-	case "baseline":
-		results = benchmarkBaseline()
+	case "baseline-structured":
+		results = benchmarkBaseline(benchmarkCases, MockedStructuredRequest)
+		break
+	case "baseline-binary":
+		results = benchmarkBaseline(benchmarkCases, MockedBinaryRequest)
+		break
+	case "binding-structured-to-structured":
+		results = benchmarkReceiverSender(benchmarkCases, MockedStructuredRequest, http.ForceStructured())
+		break
+	case "binding-structured-to-binary":
+		results = benchmarkReceiverSender(benchmarkCases, MockedStructuredRequest, http.ForceBinary())
+		break
+	case "binding-binary-to-structured":
+		results = benchmarkReceiverSender(benchmarkCases, MockedBinaryRequest, http.ForceStructured())
 		break
-	case "receiver-sender":
-		results = benchmarkReceiverSender()
+	case "binding-binary-to-binary":
+		results = benchmarkReceiverSender(benchmarkCases, MockedBinaryRequest, http.ForceBinary())
 		break
-	case "client":
-		results = benchmarkClient()
+	case "client-binary":
+		results = benchmarkClient(benchmarkCases, MockedBinaryRequest)
+		break
+	case "client-structured":
+		results = benchmarkClient(benchmarkCases, MockedStructuredRequest)
 		break
 	default:
 		panic("Wrong bench flag")
@@ -229,16 +251,17 @@ func main() {
 		_ = pprof.WriteHeapProfile(f)
 	}
 
-	writer := csv.NewWriter(os.Stdout)
+	f, err := os.Create(*out)
+	if err != nil {
+		panic(fmt.Sprintf("Cannot open file %s: %v", *out, err))
+	}
+	defer f.Close()
+
+	writer := csv.NewWriter(f)
 	defer writer.Flush()
 
-	for _, res := range results {
-		_ = writer.Write([]string{
-			strconv.Itoa(res.parallelism),
-			strconv.Itoa(res.payloadSizeKb),
-			strconv.FormatInt(res.NsPerOp(), 10),
-			strconv.FormatInt(res.AllocedBytesPerOp(), 10),
-		})
+	err = results.WriteToCsv(writer)
+	if err != nil {
+		panic(err)
 	}
-
 }
diff --git a/test/benchmark/http/plot_output_senders_allocs.gnuplot b/test/benchmark/http/plot_output_senders_allocs.gnuplot
new file mode 100644
index 0000000..d141b9e
--- /dev/null
+++ b/test/benchmark/http/plot_output_senders_allocs.gnuplot
@@ -0,0 +1,19 @@
+set datafile separator comma
+set datafile missing NaN
+set xlabel "Number of output senders"
+set ylabel "Memory Allocated/Ops"
+payload_size_kb=ARG1
+payload_size=payload_size_kb*1024
+parallelism=(exist("ARG2") && ARG2 != ""?ARG2:1)
+
+print "Plotting with payload size ".payload_size." and parallelism ".parallelism.""
+
+plot "baseline-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Baseline Binary ".payload_size_kb."kb" with linespoint, \
+     "baseline-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Baseline Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Binding Structured to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Binding Structured to Binary ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Binding Binary to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Binding Binary to Binary ".payload_size_kb."kb" with linespoint, \
+     "client-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Client Binary ".payload_size_kb."kb" with linespoint, \
+     "client-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$5:1/0) title "Client Structured ".payload_size_kb."kb" with linespoint
+pause -1
diff --git a/test/benchmark/http/plot_output_senders_ns.gnuplot b/test/benchmark/http/plot_output_senders_ns.gnuplot
new file mode 100644
index 0000000..221dd24
--- /dev/null
+++ b/test/benchmark/http/plot_output_senders_ns.gnuplot
@@ -0,0 +1,19 @@
+set datafile separator comma
+set datafile missing NaN
+set xlabel "Number of output senders"
+set ylabel "Nanoseconds/Ops"
+payload_size_kb=ARG1
+payload_size=payload_size_kb*1024
+parallelism=(exist("ARG2") && ARG2 != ""?ARG2:1)
+
+print "Plotting with payload size ".payload_size." and parallelism ".parallelism.""
+
+plot "baseline-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Baseline Binary ".payload_size_kb."kb" with linespoint, \
+     "baseline-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Baseline Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Binding Structured to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Binding Structured to Binary ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Binding Binary to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Binding Binary to Binary ".payload_size_kb."kb" with linespoint, \
+     "client-binary.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Client Binary ".payload_size_kb."kb" with linespoint, \
+     "client-structured.csv" using 3:(($2==payload_size && $1==parallelism)?$4:1/0) title "Client Structured ".payload_size_kb."kb" with linespoint
+pause -1
diff --git a/test/benchmark/http/plot_parallelism_allocs.gnuplot b/test/benchmark/http/plot_parallelism_allocs.gnuplot
new file mode 100644
index 0000000..1a8b091
--- /dev/null
+++ b/test/benchmark/http/plot_parallelism_allocs.gnuplot
@@ -0,0 +1,19 @@
+set datafile separator comma
+set datafile missing NaN
+set xlabel "Parallelism"
+set ylabel "Memory Allocated/Ops"
+payload_size_kb=ARG1
+payload_size=payload_size_kb*1024
+output_senders=(exist("ARG2") && ARG2 != ""?ARG2:1)
+
+print "Plotting with payload size ".payload_size." and output_senders ".output_senders.""
+
+plot "baseline-binary.csv" using 1:(($2==payload_size && $3==output_senders)?$5:1/0) title "Baseline Binary ".payload_size_kb."kb" with linespoint, \
+     "baseline-structured.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Baseline Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-structured.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Binding Structured to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-binary.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Binding Structured to Binary ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-structured.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Binding Binary to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-binary.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Binding Binary to Binary ".payload_size_kb."kb" with linespoint, \
+     "client-binary.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Client Binary ".payload_size_kb."kb" with linespoint, \
+     "client-structured.csv" using 1:($2==payload_size && $3==output_senders?$5:1/0) title "Client Structured ".payload_size_kb."kb" with linespoint
+pause -1
diff --git a/test/benchmark/http/plot_parallelism_ns.gnuplot b/test/benchmark/http/plot_parallelism_ns.gnuplot
index 8735dba..cdf5e42 100644
--- a/test/benchmark/http/plot_parallelism_ns.gnuplot
+++ b/test/benchmark/http/plot_parallelism_ns.gnuplot
@@ -2,6 +2,18 @@ set datafile separator comma
 set datafile missing NaN
 set xlabel "Parallelism"
 set ylabel "Nanoseconds/Ops"
-payload_size=ARG1
-plot "baseline.csv" using 1:($2==payload_size?$3:1/0) title "Baseline ".payload_size."kb" with linespoint, "receiver-sender.csv" using 1:($2==payload_size?$3:1/0) title "Receiver Sender ".payload_size."kb" with linespoint, "pipe.csv" using 1:($2==payload_size?$3:1/0) title "Pipe ".payload_size."kb" with linespoint, "client.csv" using 1:($2==payload_size?$3:1/0) title "Client ".payload_size."kb" with linespoint
+payload_size_kb=ARG1
+payload_size=payload_size_kb*1024
+output_senders=(exist("ARG2") && ARG2 != ""?ARG2:1)
+
+print "Plotting with payload size ".payload_size." and output_senders ".output_senders.""
+
+plot "baseline-binary.csv" using 1:(($2==payload_size && $3==output_senders)?$4:1/0) title "Baseline Binary ".payload_size_kb."kb" with linespoint, \
+     "baseline-structured.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Baseline Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-structured.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Binding Structured to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-structured-to-binary.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Binding Structured to Binary ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-structured.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Binding Binary to Structured ".payload_size_kb."kb" with linespoint, \
+     "binding-binary-to-binary.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Binding Binary to Binary ".payload_size_kb."kb" with linespoint, \
+     "client-binary.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Client Binary ".payload_size_kb."kb" with linespoint, \
+     "client-structured.csv" using 1:($2==payload_size && $3==output_senders?$4:1/0) title "Client Structured ".payload_size_kb."kb" with linespoint
 pause -1
diff --git a/test/benchmark/http/run_revision.sh b/test/benchmark/http/run_revision.sh
new file mode 100755
index 0000000..b6a282f
--- /dev/null
+++ b/test/benchmark/http/run_revision.sh
@@ -0,0 +1,87 @@
+#!/bin/bash
+
+set -e
+
+RUNNABLE_NAME="http-bench"
+
+function usage {
+   echo "Usage: $0 [--cpu-profile] [--mem-profile] [--max-parallelism n] [--max-payload n] [--max-output-senders n] [git_revision]"
+   exit 1
+}
+
+ADDITIONAL_ARGS=""
+
+PARAMS=""
+while (( "$#" )); do
+  case "$1" in
+    -h|--help)
+      usage
+      ;;
+    -c|--cpu-profile)
+      CPU_PROFILE="1"
+      shift
+      ;;
+    -m|--mem-profile)
+      MEM_PROFILE="1"
+      shift
+      ;;
+    --max-parallelism)
+      ADDITIONAL_ARGS="$ADDITIONAL_ARGS --max-parallelism $2"
+      shift 2
+      ;;
+    --max-payload)
+      ADDITIONAL_ARGS="$ADDITIONAL_ARGS --max-payload $2"
+      shift 2
+      ;;
+    --max-output-senders)
+      ADDITIONAL_ARGS="$ADDITIONAL_ARGS --max-output-senders $2"
+      shift 2
+      ;;
+    --) # end argument parsing
+      shift
+      break
+      ;;
+    -*|--*=) # unsupported flags
+      echo "Error: Unsupported flag $1" >&2
+      exit 1
+      ;;
+    *) # preserve positional arguments
+      PARAMS="$PARAMS $1"
+      shift
+      ;;
+  esac
+done
+eval set -- "$PARAMS"
+
+REVISION=$1
+if [ ! -z "$REVISION" ]
+then
+      git checkout "$REVISION"
+else
+      REVISION=results
+fi
+
+go build -o $RUNNABLE_NAME -v github.com/cloudevents/sdk-go/test/benchmark/http
+
+mkdir -p "$REVISION"
+
+BENCHS=(
+  "baseline-structured"
+  "baseline-binary"
+  "binding-structured-to-structured"
+  "binding-structured-to-binary"
+  "binding-binary-to-structured"
+  "binding-binary-to-binary"
+  "client-binary"
+  "client-structured"
+)
+
+for i in "${BENCHS[@]}"; do
+    ./$RUNNABLE_NAME --bench="$i" \
+    ${MEM_PROFILE+-memprofile $REVISION/$i-mem.pprof} \
+    ${CPU_PROFILE+-cpuprofile $REVISION/$i-cpu.pprof} \
+    $ADDITIONAL_ARGS \
+    --out="$REVISION/$i.csv"
+done
+
+rm $RUNNABLE_NAME