*: cpnp tweaks

cmd/thanos/receive.go

@@ -464,7 +464,7 @@ func runReceive(
 		if err != nil {
 			return err
 		}
-		server := receive.NewCapNProtoServer(listener, handler)
+		server := receive.NewCapNProtoServer(listener, handler, logger)
 		g.Add(func() error {
 			return server.ListenAndServe()
 		}, func(err error) {

pkg/receive/capnp_server.go

@@ -13,18 +13,28 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/pkg/errors"
 
+	"github.com/thanos-io/thanos/pkg/pool"
 	"github.com/thanos-io/thanos/pkg/receive/writecapnp"
+	"github.com/thanos-io/thanos/pkg/runutil"
 )
 
 type CapNProtoServer struct {
 	listener net.Listener
 	server   writecapnp.Writer
+	logger   log.Logger
+	workers  pool.WorkerPool
 }
 
-func NewCapNProtoServer(listener net.Listener, handler *CapNProtoHandler) *CapNProtoServer {
+func NewCapNProtoServer(listener net.Listener, handler *CapNProtoHandler, logger log.Logger) *CapNProtoServer {
+	const cpnpWorkerPoolSize = 512
+
 	return &CapNProtoServer{
 		listener: listener,
 		server:   writecapnp.Writer_ServerToClient(handler),
+		logger:   logger,
+		workers: pool.NewWorkerPool(
+			cpnpWorkerPoolSize,
+		),
 	}
 }
 
@@ -34,15 +44,16 @@ func (c *CapNProtoServer) ListenAndServe() error {
 		if err != nil {
 			return err
 		}
-		go func() {
-			defer conn.Close()
+
+		c.workers.Go(func() {
+			defer runutil.CloseWithLogOnErr(c.logger, conn, "receive capnp conn")
 			rpcConn := rpc.NewConn(rpc.NewPackedStreamTransport(conn), &rpc.Options{
 				// The BootstrapClient is the RPC interface that will be made available
 				// to the remote endpoint by default.
 				BootstrapClient: capnp.Client(c.server).AddRef(),
 			})
 			<-rpcConn.Done()
-		}()
+		})
 	}
 }
 

pkg/receive/capnproto_server_bench_test.go

@@ -54,7 +54,7 @@ func BenchmarkCapNProtoServer_SingleConcurrentClient(b *testing.B) {
 		)
 		listener = bufconn.Listen(1024)
 		handler  = NewCapNProtoHandler(log.NewNopLogger(), writer)
-		srv      = NewCapNProtoServer(listener, handler)
+		srv      = NewCapNProtoServer(listener, handler, log.NewNopLogger())
 	)
 	go func() {
 		_ = srv.ListenAndServe()

pkg/receive/capnproto_server_test.go

@@ -26,7 +26,7 @@ func TestCapNProtoServer_SingleConcurrentClient(t *testing.T) {
 		)
 		listener = bufconn.Listen(1024)
 		handler  = NewCapNProtoHandler(log.NewNopLogger(), writer)
-		srv      = NewCapNProtoServer(listener, handler)
+		srv      = NewCapNProtoServer(listener, handler, log.NewNopLogger())
 	)
 	go func() {
 		_ = srv.ListenAndServe()
@@ -54,7 +54,7 @@ func TestCapNProtoServer_MultipleConcurrentClients(t *testing.T) {
 		)
 		listener = bufconn.Listen(1024)
 		handler  = NewCapNProtoHandler(log.NewNopLogger(), writer)
-		srv      = NewCapNProtoServer(listener, handler)
+		srv      = NewCapNProtoServer(listener, handler, log.NewNopLogger())
 	)
 	go func() {
 		_ = srv.ListenAndServe()

pkg/receive/handler_test.go

@@ -22,7 +22,7 @@
 	"testing"
 	"time"
 
 	goerrors "errors"
 
 	"github.com/alecthomas/units"
 	"github.com/go-kit/log"
@@ -256,7 +256,7 @@
 				listener = bufconn.Listen(1024)
 				handler  = NewCapNProtoHandler(log.NewNopLogger(), writer)
 			)
-			srv := NewCapNProtoServer(listener, handler)
+			srv := NewCapNProtoServer(listener, handler, log.NewNopLogger())
 			client := writecapnp.NewRemoteWriteClient(listener, logger)
 			peer = newPeerWorker(client, prometheus.NewHistogram(prometheus.HistogramOpts{}), 1)
 			closers = append(closers, func() error {

pkg/receive/writecapnp/write_request.go

@@ -4,6 +4,8 @@
 package writecapnp
 
 import (
+	"unsafe"
+
 	"github.com/prometheus/prometheus/model/exemplar"
 	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
@@ -59,7 +61,9 @@ func NewRequest(wr WriteRequest) (*Request, error) {
 	start := uint32(0)
 	for i := 0; i < offsets.Len(); i++ {
 		end := offsets.At(i)
-		*strings = append(*strings, string(data[start:end]))
+
+		b := data[start:end]
+		*strings = append(*strings, unsafe.String(&b[0], len(b)))
 		start = end
 	}
 

pkg/receive/writecapnp/write_request_test.go

@@ -0,0 +1,39 @@
+package writecapnp
+
+import (
+	"testing"
+
+	"capnproto.org/go/capnp/v3"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewRequest(t *testing.T) {
+	arena := capnp.SingleSegment(nil)
+
+	_, seg, err := capnp.NewMessage(arena)
+	require.NoError(t, err)
+
+	wr, err := NewRootWriteRequest(seg)
+	require.NoError(t, err)
+
+	symbols, err := NewSymbols(seg)
+	require.NoError(t, err)
+
+	symbols.SetData([]byte(`foobar`))
+	list, err := capnp.NewUInt32List(
+		seg, 2,
+	)
+	require.NoError(t, err)
+	list.Set(0, 3)
+	list.Set(1, 6)
+
+	symbols.SetOffsets(list)
+	wr.SetSymbols(symbols)
+
+	req, err := NewRequest(wr)
+	require.NoError(t, err)
+
+	require.Equal(t, "foo", (*req.symbols)[0])
+	require.Equal(t, "bar", (*req.symbols)[1])
+
+}