qrpc

package module
v0.0.0-...-1bf0eed Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 14, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

README

Русская версия

qrpc

Go Version License

qrpc — a high-performance RPC framework built on QUIC (HTTP/3 transport). It uses protocol buffers with vtprotobuf acceleration, lock-free stream multiplexing, and aggressive object pooling to deliver low-latency, high-throughput communication. Supports both request-response RPC and one-way event messaging.

Features

  • QUIC Transport — UDP + TLS 1.3, no head-of-line blocking, built-in encryption.
  • Protobuf + vtprotobuf — compact binary serialization with zero-copy marshal.
  • Stream Multiplexing — N pre-opened QUIC streams per connection, lock-free atomic round-robin balancer.
  • Object Pooling — sync.Pool for encoder buffers, request/response objects, and response channels — minimizes GC pressure.
  • Sharded Concurrent Map — 256-shard map for O(1) request-ID-to-channel dispatch.
  • Event Messaging — one-way event delivery alongside request-response RPC.
  • Simple API — NewServer → AddHandler / AddEventHandler, NewClient → NewRequest → SendRequest / SendEvent.

Quick Start

Server
package main

import (
  "crypto/tls"
  "log"
  "github.com/XeshSufferer/qrpc"
)

func main() {
  tlsConfig := /* *tls.Config */
  server, err := qrpc.NewServer("0.0.0.0:8081", tlsConfig)
  if err != nil {
    log.Fatal(err)
  }
  server.AddHandler("echo", func(c qrpc.Ctx) {
    c.SetBody(c.Body())
    c.SetCode(0)
  })
  select {}
}
Client
package main

import (
  "context"
  "crypto/tls"
  "log"
  "github.com/XeshSufferer/qrpc"
)

func main() {
  tlsConfig := /* *tls.Config */
  client, err := qrpc.NewClient(context.Background(), "127.0.0.1:8081", tlsConfig, 1)
  if err != nil {
    log.Fatal(err)
  }

  req := client.NewRequest()
  req.SetMethod([]byte("echo"))
  req.SetBody([]byte("hello qrpc"))

  resp, err := client.SendRequest(context.Background(), req)
  if err != nil {
    log.Fatal(err)
  }
  log.Printf("Response: %s", resp.Body())
  client.ReleaseResponse(resp)
}
Events (one-way messaging)
// Server
server.AddEventHandler("notify", func(c qrpc.EventCtx) {
  log.Printf("event %s: %s", c.Method(), c.Body())
})

// Client
req := client.NewRequest()
req.SetMethod([]byte("notify"))
req.SetBody([]byte("hello"))
err := client.SendEvent(context.Background(), req)

Wire Protocol

Frame Format
┌─────────────────────────────────────────────┐
│  4 bytes: payload length (big-endian uint32) │
├─────────────────────────────────────────────┤
│  1 byte:  flag                               │
│           REQUEST       = 1                  │
│           RESPONSE      = 2                  │
│           EVENT         = 3                  │
│           REQUEST_ZSTD  = 4                  │
│           RESPONSE_ZSTD = 5                  │
│           EVENT_ZSTD    = 6                  │
├─────────────────────────────────────────────┤
│  N bytes: protobuf Request / Response        │
└─────────────────────────────────────────────┘
  • payload length = 1 (flag) + protobuf bytes
  • vtprotobuf uses MarshalToSizedBufferVT (reverse write into pre-allocated buffer)
Messages (protobuf)
message Request {
  uint64 request_id = 3;
  bytes  headers    = 1;
  bytes  method     = 2;
  bytes  body       = 4;
}

message Response {
  uint64 request_id = 3;
  uint32 code       = 5;
  bytes  headers    = 1;
  bytes  method     = 2;
  bytes  body       = 4;
}

Requests are matched to responses via a random uint64 request_id. Events use request_id = 0 and do not generate a response.

Payloads larger than 16 KB are automatically compressed with zstd (flags 4–6).

Architecture

┌───────────────┐     QUIC (UDP)      ┌───────────────┐
│   Client      │ ◄─────────────────► │   Server      │
│               │  TLS 1.3, ALPN     │               │
│  ┌─────────┐  │  "qrpc"            │  ┌─────────┐  │
│  │ Sharded │  │                     │  │Handlers │  │
│  │  Map    │  │                     │  │  Map    │  │
│  └────┬────┘  │                     │  └─────────┘  │
│  ┌────▼────┐  │  N pre-opened      │  ┌─────────┐  │
│  │Multiplex│  │  streams            │  │  Stream │  │
│  │   -er   │──┤───────────────────►│  │  Read   │  │
│  └────┬────┘  │                     │  │  Cycle  │  │
│  ┌────▼────┐  │                     │  └─────────┘  │
│  │Balancer │  │  atomic round-robin │               │
│  │(lockfree)│  │                     │               │
│  └─────────┘  │                     │               │
└───────────────┘                     └───────────────┘

Client flow. On NewClient, the multiplexer opens N QUIC streams (default 32) per connection and starts a read-cycle goroutine per stream. NewRequest obtains a pooled Request wrapping a protobuf Request. SendRequest assigns a random request_id, stores a chan *Response in the sharded map, encodes the request via the encoder, and writes it to a stream obtained from the round-robin balancer. The read-cycle goroutine decodes incoming frames, looks up request_id in the sharded map, and dispatches the response to the waiting channel. SendEvent writes a one-way frame with request_id = 0 and no response is expected.

Server flow. NewServer starts a QUIC listener. Each accepted connection gets a goroutine that accepts streams. Each stream runs a read cycle that decodes frames. Request frames (flag 1/4) dispatch to the registered handler via qrpc.Ctx; the handler sets the response and the server encodes and writes it back. Event frames (flag 3/6) dispatch to the event handler via qrpc.EventCtx; no response is sent.

API Reference

Server
func NewServer(addr string, tls *tls.Config) (QRpcServer, error)
Method Signature Description
AddHandler (method string, handler func(qrpc.Ctx)) Register an RPC handler
AddEventHandler (method string, handler func(qrpc.EventCtx)) Register an event handler

qrpc.Ctx (RPC handler):

Method Returns Description
Body() []byte Request body
Headers() [][]byte Request headers (key-value pairs)
Method() []byte Request method name
GetHeader(key, default) string Get a header value by key
SetBody([]byte) — Set response body
SetCode(uint32) — Set response status code
SetHeader(key, value) — Set a response header
SetHeaders([][]byte) — Set all response headers
Locals() Locals Per-request local storage

qrpc.EventCtx (event handler): Body(), Headers(), Method(), GetHeader(), Locals() — read-only, no response.

Client
func NewClient(ctx context.Context, addr string, tls *tls.Config, connsCount int) (Client, error)
Method Returns Description
NewRequest() Request Get a pooled request context
SendRequest(ctx, req) (Response, error) Send RPC and wait for response
SendEvent(ctx, req) error Fire-and-forget event
ReleaseResponse(Response) — Return response to pool
ReleaseRequest(Request) — Return request to pool

Request: Body(), SetBody(), Headers(), SetHeaders(), Method(), SetMethod(), RequestId(), Locals().

Response: Body(), Headers(), Code(), RequestId().

Performance

Benchmarks run on localhost with Linux tc netem for network emulation. All runs test 5 network profiles × 5 scenarios × 2 payload sizes (100 B / 4 KB) × 2 connection counts (1 / 16), using 64 pipelining, 16 QUIC streams, and 10s duration + 3s warmup.

Throughput (RPS) — Best Across Payload Sizes & Connection Counts
Scenario Workers clean wifi lte bad_lte extreme
baseline_latency 1 23,607 3,690 625 278 102
concurrency_stress 100 506,590 60,334 5,355 1,576 827
multiplex_stress 50 478,703 60,234 4,864 1,229 720
loss_sensitivity 100 484,582 59,982 4,568 1,207 619
rtt_scaling 50 462,074 60,467 5,316 1,339 831

All scenarios maintain 100% success rate across all network profiles.

Tail Latency (P95) — Worst Across Payload Sizes & Connection Counts
Scenario clean wifi lte bad_lte extreme
baseline_latency 3.4 ms 23.5 ms 405 ms 1.74 s 2.85 s
concurrency_stress 1.74 s 7.69 s 8.74 s 8.46 s 8.46 s
multiplex_stress 23.8 ms 5.04 s 5.90 s 4.84 s 7.07 s
loss_sensitivity 1.60 s 4.89 s 4.99 s 7.24 s 6.49 s
rtt_scaling 175 ms 5.08 s 5.39 s 6.13 s 6.68 s

On clean networks qrpc delivers 23K–506K RPS with low millisecond latency. Under extreme conditions (150ms RTT, 10% loss) throughput degrades predictably while maintaining 100% delivery — 102–831 RPS with P95 latency of 2.85–8.46 s depending on concurrency.

Installation

go get github.com/XeshSufferer/qrpc

Requires Go 1.26.2+.

Benchmarking

Unit Benchmarks
go test -bench=. -benchmem ./...
Stress Test Framework
cd stress_tester
go build -o stress_tester .

# Start a benchmark server
sudo ./stress_tester server -addr 127.0.0.1:8081

# Run a specific scenario
sudo ./stress_tester run -scenario baseline_latency -profile clean,wifi,lte -system qrpc -addr 127.0.0.1:8081

# Run the full automated suite
sudo ./run_all.sh --duration 10s --warmup 3s
Automated Suite (100 runs)
go run ./runall_stress/main.go

Runs all 5 scenarios × 5 profiles × 2 payload sizes × 2 connection counts and generates results/heatmap.html.

HTML Report
cd stress_tester
python3 analyze.py results --html report.html

Network profiles are applied via Linux tc netem (requires sudo).

Network Profiles

Profile RTT Jitter Loss Bandwidth Description
clean 1ms 0ms 0% 1000 Mbps Local / DC
wifi 5ms 2ms 0.1% 100 Mbps Typical WiFi
lte 30ms 10ms 1% 50 Mbps 4G LTE
bad_lte 60ms 20ms 5% 10 Mbps Poor LTE
extreme 150ms 50ms 10% 5 Mbps Extreme conditions

Test Scenarios

Scenario Workers Streams Pipelining Payload Profiles Used
baseline_latency 1 16 64 100 B / 4 KB fixed clean, wifi, lte, bad_lte, extreme
concurrency_stress 100 16 64 100 B / 4 KB fixed clean, wifi, lte, bad_lte, extreme
multiplex_stress 50 16 64 100 B / 4 KB fixed clean, wifi, lte, bad_lte, extreme
loss_sensitivity 100 16 64 100 B / 4 KB fixed clean, wifi, lte, bad_lte, extreme
rtt_scaling 50 16 64 100 B / 4 KB fixed clean, wifi, lte, bad_lte, extreme

Dependencies

Library Version Purpose
quic-go v0.59.1 QUIC transport
vtprotobuf v0.6.0 Fast protobuf marshal
google.golang.org/protobuf v1.36.11 Protobuf runtime

License

Apache License 2.0 — see LICENSE.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (

	// ErrClientClosed is returned when attempting to use a closed client.
	ErrClientClosed = errors.New("client is closed")
)
View Source 
var TimeoutDuration = time.Second * 30

TimeoutDuration is the write deadline applied to stream operations. It defaults to 30 seconds.

Functions

func IsTimeoutErr 

func IsTimeoutErr(err error) bool

IsTimeoutErr checks whether the given error is a QUIC idle timeout error.

func ReleaseCtx 

func ReleaseCtx(ctx *CtxImpl)

ReleaseCtx returns a CtxImpl to the pool for reuse.

func ReleaseRequest 

func ReleaseRequest(ctx *RequestImpl)

ReleaseRequest returns a RequestImpl to the pool for reuse.

func ReleaseResponse 

func ReleaseResponse(ctx *ResponseImpl)

ReleaseResponse returns a ResponseImpl to the pool for reuse.

Types

type Client 

type Client interface {
	NewRequest() Request
	SendRequest(ctx context.Context, req Request) (Response, error)
	ReleaseResponse(resp Response)
	SendEvent(ctx context.Context, req Request) error
	ReleaseRequest(req Request)
	Close()
}

Client is the qrpc client interface for sending RPC requests and one-way events over QUIC connections. Implementations use stream multiplexing, object pooling, and a sharded concurrent map for request-response dispatch.

func NewClient 

func NewClient(ctx context.Context, addr string, tls *tls.Config, connsCount int) (Client, error)

NewClient creates a new qrpc client with connsCount QUIC connections to the given addr. Each connection opens a pool of pre-created QUIC streams and runs read-cycle goroutines.

type ClientImpl 

type ClientImpl struct {
	// contains filtered or unexported fields
}

ClientImpl is the concrete implementation of Client. It manages multiple QUIC connections, stream multiplexers, and response channels.

func (*ClientImpl) Close 

func (c *ClientImpl) Close()

Close gracefully shuts down the client, waits for pending requests, and closes all multiplexers and QUIC connections.

func (*ClientImpl) NewRequest 

func (c *ClientImpl) NewRequest() Request

NewRequest obtains a pooled Request wrapping a protobuf Request message.

func (*ClientImpl) ReleaseRequest 

func (c *ClientImpl) ReleaseRequest(req Request)

ReleaseRequest returns a Request to the pool for reuse.

func (*ClientImpl) ReleaseResponse 

func (c *ClientImpl) ReleaseResponse(resp Response)

ReleaseResponse returns a Response to the pool for reuse.

func (*ClientImpl) SendEvent 

func (c *ClientImpl) SendEvent(
	ctx context.Context,
	req Request,
) error

SendEvent sends a one-way event with no response expected. The request is released back to the pool after sending.

func (*ClientImpl) SendRequest 

func (c *ClientImpl) SendRequest(
	ctx context.Context,
	req Request,
) (Response, error)

SendRequest sends an RPC request and waits for a response. The request is released back to the pool after sending.

type Ctx 

type Ctx interface {
	Locals() Locals
	Body() []byte
	Headers() [][]byte
	Method() []byte
	SetBody(buff []byte)
	SetHeaders(buff [][]byte)
	GetHeader(key, defaultValue string) string
	SetHeader(key, value string)
	SetCode(code uint32)
}

Ctx is the context passed to RPC handlers. It provides read access to the incoming request and write access to the response.

type CtxImpl 

type CtxImpl struct {
	// contains filtered or unexported fields
}

CtxImpl is the concrete implementation of Ctx and EventCtx. It wraps a protobuf Request and Response and is pooled via sync.Pool.

func NewCtx 

func NewCtx(req *gen.Request, resp *gen.Response) *CtxImpl

NewCtx obtains a pooled CtxImpl wrapping the given request and response.

func (*CtxImpl) Body 

func (c *CtxImpl) Body() []byte

func (*CtxImpl) GetHeader 

func (c *CtxImpl) GetHeader(key, defaultValue string) string

func (*CtxImpl) Headers 

func (c *CtxImpl) Headers() [][]byte

func (*CtxImpl) Locals 

func (c *CtxImpl) Locals() Locals

func (*CtxImpl) Method 

func (c *CtxImpl) Method() []byte

func (*CtxImpl) SetBody 

func (c *CtxImpl) SetBody(buff []byte)

func (*CtxImpl) SetCode 

func (c *CtxImpl) SetCode(code uint32)

func (*CtxImpl) SetHeader 

func (c *CtxImpl) SetHeader(key, value string)

func (*CtxImpl) SetHeaders 

func (c *CtxImpl) SetHeaders(buff [][]byte)

type EventCtx 

type EventCtx interface {
	Locals() Locals
	Body() []byte
	Headers() [][]byte
	GetHeader(key, defaultValue string) string
	Method() []byte
}

EventCtx is the context passed to event handlers. It provides read-only access to the event — no response is sent.

type Locals 

type Locals interface {
	SetString(key, value string)
	GetString(key string) string
	Set(key string, value any)
	Get(key string) any
	Reset()
}

Locals is a per-request local storage for passing arbitrary data between middlewares and handlers.

func NewLocals 

func NewLocals() Locals

NewLocals creates a new empty Locals.

type LocalsImpl 

type LocalsImpl struct {
	// contains filtered or unexported fields
}

LocalsImpl is the concrete implementation of Locals with separate maps for string and arbitrary values, protected by an RWMutex.

func (*LocalsImpl) Get 

func (l *LocalsImpl) Get(key string) any

func (*LocalsImpl) GetString 

func (l *LocalsImpl) GetString(key string) string

func (*LocalsImpl) Reset 

func (l *LocalsImpl) Reset()

func (*LocalsImpl) Set 

func (l *LocalsImpl) Set(key string, value any)

func (*LocalsImpl) SetString 

func (l *LocalsImpl) SetString(key, value string)

type QRPCServerImpl 

type QRPCServerImpl struct {
	// contains filtered or unexported fields
}

QRPCServerImpl is the concrete implementation of QRpcServer. It listens on a QUIC address, accepts streams, and dispatches requests and events to the registered handlers.

func (*QRPCServerImpl) AddEventHandler 

func (s *QRPCServerImpl) AddEventHandler(method string, handler func(EventCtx))

AddEventHandler registers a one-way event handler for the given method name.

func (*QRPCServerImpl) AddHandler 

func (s *QRPCServerImpl) AddHandler(method string, handler func(Ctx))

AddHandler registers an RPC handler for the given method name.

type QRpcServer 

type QRpcServer interface {
	AddHandler(method string, handler func(Ctx))
	AddEventHandler(method string, handler func(EventCtx))
	// contains filtered or unexported methods
}

QRpcServer is the server interface for registering RPC and event handlers and accepting QUIC connections.

func NewServer 

func NewServer(addr string, tls *tls.Config) (QRpcServer, error)

NewServer creates a new QUIC-based RPC server, starts listening on addr, and returns immediately. The server runs in the background.

type Request 

type Request interface {
	Locals() Locals
	Body() []byte
	SetBody([]byte)
	Headers() [][]byte
	SetHeaders([][]byte)
	Method() []byte
	SetMethod([]byte)
	RequestId() uint64
}

Request is the client-side request interface for setting method, body, headers, and accessing per-request local storage.

type RequestImpl 

type RequestImpl struct {
	// contains filtered or unexported fields
}

RequestImpl wraps a protobuf Request and provides the Request interface.

func NewRequest 

func NewRequest(req *gen.Request) *RequestImpl

NewRequest wraps a protobuf Request in a pooled RequestImpl.

func (*RequestImpl) Body 

func (c *RequestImpl) Body() []byte

func (*RequestImpl) Headers 

func (c *RequestImpl) Headers() [][]byte

func (*RequestImpl) Locals 

func (c *RequestImpl) Locals() Locals

func (*RequestImpl) Method 

func (c *RequestImpl) Method() []byte

func (*RequestImpl) Req 

func (c *RequestImpl) Req() *gen.Request

func (*RequestImpl) RequestId 

func (c *RequestImpl) RequestId() uint64

func (*RequestImpl) SetBody 

func (c *RequestImpl) SetBody(b []byte)

func (*RequestImpl) SetHeaders 

func (c *RequestImpl) SetHeaders(h [][]byte)

func (*RequestImpl) SetMethod 

func (c *RequestImpl) SetMethod(m []byte)

type Response 

type Response interface {
	Body() []byte
	Headers() [][]byte
	Code() uint32
	RequestId() uint64
}

Response is the client-side response interface for reading the response body, headers, status code, and matching request ID.

type ResponseImpl 

type ResponseImpl struct {
	// contains filtered or unexported fields
}

ResponseImpl wraps a protobuf Response and provides the Response interface.

func NewResponse 

func NewResponse(resp *gen.Response) *ResponseImpl

NewResponse wraps a protobuf Response in a pooled ResponseImpl.

func (*ResponseImpl) Body 

func (c *ResponseImpl) Body() []byte

func (*ResponseImpl) Code 

func (c *ResponseImpl) Code() uint32

func (*ResponseImpl) Headers 

func (c *ResponseImpl) Headers() [][]byte

func (*ResponseImpl) RequestId 

func (c *ResponseImpl) RequestId() uint64

func (*ResponseImpl) Resp 

func (c *ResponseImpl) Resp() *gen.Response

Source Files

View all Source files

Directories

Path Synopsis
protos
pb/gen
transport
quic
quic/client
types

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.