test: compare native concurrency and server memory
This commit is contained in:
125
transport/benchmarkTCP_test.go
Normal file
125
transport/benchmarkTCP_test.go
Normal file
@@ -0,0 +1,125 @@
|
||||
package transport_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
omni "git.yuntuo.hk/k2safe/go-omni"
|
||||
"git.yuntuo.hk/k2safe/go-omni/transport/native"
|
||||
)
|
||||
|
||||
// Both servers use the same application, client, request validation and fixed
|
||||
// concurrency. B/op includes client AND server allocations in this process;
|
||||
// ns/op is aggregate throughput cost, not individual request latency.
|
||||
func BenchmarkTransportTCPApplicationRequest(b *testing.B) {
|
||||
for _, adapter := range []string{"nethttp", "native"} {
|
||||
for _, concurrency := range []int{1, 32, 128} {
|
||||
b.Run(fmt.Sprintf("%s/concurrency%d", adapter, concurrency), func(b *testing.B) {
|
||||
benchmarkTCP(b, adapter, concurrency)
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func benchmarkTCP(b *testing.B, adapter string, concurrency int) {
|
||||
application := omni.New()
|
||||
application.GET("/health", func(ctx *omni.Context) error {
|
||||
return ctx.OK(map[string]string{"status": "ok"})
|
||||
})
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
var shutdown func(context.Context) error
|
||||
if adapter == "native" {
|
||||
server := native.New(application)
|
||||
server.MaxConnections = 1024
|
||||
server.MaxConcurrentRequests = 1024
|
||||
shutdown = server.Shutdown
|
||||
go func() { done <- server.Serve(context.Background(), listener) }()
|
||||
} else {
|
||||
server := &http.Server{
|
||||
Handler: application, ReadHeaderTimeout: 5 * time.Second,
|
||||
ReadTimeout: 30 * time.Second, WriteTimeout: 30 * time.Second,
|
||||
IdleTimeout: time.Minute, MaxHeaderBytes: 1 << 20,
|
||||
}
|
||||
shutdown = server.Shutdown
|
||||
go func() {
|
||||
err := server.Serve(listener)
|
||||
if err == http.ErrServerClosed {
|
||||
err = nil
|
||||
}
|
||||
done <- err
|
||||
}()
|
||||
}
|
||||
client := &http.Client{Transport: &http.Transport{
|
||||
MaxIdleConns: 1024,
|
||||
MaxIdleConnsPerHost: 1024,
|
||||
MaxConnsPerHost: concurrency,
|
||||
IdleConnTimeout: time.Minute,
|
||||
}, Timeout: 5 * time.Second}
|
||||
b.Cleanup(func() {
|
||||
client.CloseIdleConnections()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := shutdown(ctx); err != nil {
|
||||
b.Error(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
b.Error(err)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
b.Error("server did not stop")
|
||||
}
|
||||
})
|
||||
endpoint := "http://" + listener.Addr().String() + "/health"
|
||||
request := func() error {
|
||||
response, err := client.Get(endpoint)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, readErr := io.Copy(io.Discard, response.Body)
|
||||
closeErr := response.Body.Close()
|
||||
if response.StatusCode != http.StatusOK || response.Header.Get("X-Request-ID") == "" {
|
||||
return fmt.Errorf("unexpected response: status=%d", response.StatusCode)
|
||||
}
|
||||
if readErr != nil {
|
||||
return readErr
|
||||
}
|
||||
return closeErr
|
||||
}
|
||||
if err := request(); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
errors := make(chan error, concurrency)
|
||||
var workers sync.WaitGroup
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for worker := 0; worker < concurrency; worker++ {
|
||||
workers.Add(1)
|
||||
go func(worker int) {
|
||||
defer workers.Done()
|
||||
for index := worker; index < b.N; index += concurrency {
|
||||
if err := request(); err != nil {
|
||||
errors <- err
|
||||
return
|
||||
}
|
||||
}
|
||||
}(worker)
|
||||
}
|
||||
workers.Wait()
|
||||
b.StopTimer()
|
||||
close(errors)
|
||||
for err := range errors {
|
||||
b.Error(err)
|
||||
}
|
||||
}
|
||||
@@ -1,95 +0,0 @@
|
||||
package native_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
omni "git.yuntuo.hk/k2safe/go-omni"
|
||||
"git.yuntuo.hk/k2safe/go-omni/transport/native"
|
||||
netrouter "git.yuntuo.hk/k2safe/go-omni/transport/nethttp"
|
||||
)
|
||||
|
||||
// BenchmarkNativeApplicationRequest measures the real TCP path with a reused
|
||||
// client connection. It is intentionally separate from the in-process router
|
||||
// benchmarks in transport/benchmark_test.go.
|
||||
func BenchmarkNativeApplicationRequest(b *testing.B) {
|
||||
application := omni.NewWithRouter(netrouter.New())
|
||||
application.GET("/health", func(ctx *omni.Context) error {
|
||||
return ctx.OK(map[string]string{"status": "ok"})
|
||||
})
|
||||
server := native.New(application)
|
||||
server.MaxConnections = 1024
|
||||
server.MaxConcurrentRequests = 1024
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- server.Serve(context.Background(), listener) }()
|
||||
waitForBenchmarkServer(b, listener.Addr().String())
|
||||
|
||||
client := &http.Client{Transport: &http.Transport{
|
||||
MaxIdleConns: 1024,
|
||||
MaxIdleConnsPerHost: 1024,
|
||||
IdleConnTimeout: time.Minute,
|
||||
}}
|
||||
endpoint := "http://" + listener.Addr().String() + "/health"
|
||||
response, err := client.Get(endpoint)
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
_, _ = io.Copy(io.Discard, response.Body)
|
||||
response.Body.Close()
|
||||
|
||||
var failed atomic.Bool
|
||||
var failure atomic.Value
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
b.RunParallel(func(pb *testing.PB) {
|
||||
for pb.Next() {
|
||||
response, err := client.Get(endpoint)
|
||||
if err != nil {
|
||||
if failed.CompareAndSwap(false, true) {
|
||||
failure.Store(err.Error())
|
||||
}
|
||||
continue
|
||||
}
|
||||
_, _ = io.Copy(io.Discard, response.Body)
|
||||
response.Body.Close()
|
||||
}
|
||||
})
|
||||
b.StopTimer()
|
||||
client.CloseIdleConnections()
|
||||
shutdownContext, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
if err := server.Shutdown(shutdownContext); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
if err := <-done; err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
if failed.Load() {
|
||||
b.Fatalf("native request failed: %s", failure.Load().(string))
|
||||
}
|
||||
}
|
||||
|
||||
func waitForBenchmarkServer(b *testing.B, address string) {
|
||||
b.Helper()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
conn, err := net.DialTimeout("tcp", address, 50*time.Millisecond)
|
||||
if err == nil {
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
b.Fatalf("server did not start: %v", err)
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user