hedgedhttp

README

hedgedhttp

Hedged HTTP client which helps to reduce tail latency at scale.

Rationale

See paper Tail at Scale by Jeffrey Dean, Luiz André Barroso. In short: the client first sends one request, but then sends an additional request after a timeout if the previous hasn't returned an answer in the expected time. The client cancels remaining requests once the first result is received.

Acknowledge

Thanks to Bohdan Storozhuk for the review and powerful hints.

Features

• Simple API.
• Easy to integrate.
• Optimized for speed.
• Clean and tested code.
• Supports http.Client and http.RoundTripper.
• Dependency-free.

Install

Go version 1.16+

go get github.com/cristalhq/hedgedhttp

Example

ctx := context.Background()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://google.com", http.NoBody)
if err != nil {
	panic(err)
}

timeout := 10 * time.Millisecond
upto := 7
client := &http.Client{Timeout: time.Second}
hedged, err := hedgedhttp.NewClient(timeout, upto, client)
if err != nil {
	panic(err)
}

// will take `upto` requests, with a `timeout` delay between them
resp, err := hedged.Do(req)
if err != nil {
	panic(err)
}
defer resp.Body.Close()

Also see examples: examples_test.go.

Documentation

See these docs.

License

MIT License.

Documentation

Overview

Example (ConfigNext)

package main

import (
	"net/http"
	"sync/atomic"
	"time"

	"github.com/cristalhq/hedgedhttp"
)

func main() {
	rt := &observableRoundTripper{
		rt: http.DefaultTransport,
	}

	cfg := hedgedhttp.Config{
		Transport: rt,
		Upto:      3,
		Delay:     50 * time.Millisecond,
		Next: func() (upto int, delay time.Duration) {
			return 3, rt.MaxLatency()
		},
	}
	client, err := hedgedhttp.New(cfg)
	if err != nil {
		panic(err)
	}

	// or client.Do
	resp, err := client.RoundTrip(&http.Request{})
	_ = resp

}

type observableRoundTripper struct {
	rt         http.RoundTripper
	maxLatency atomic.Uint64
}

func (ort *observableRoundTripper) MaxLatency() time.Duration {
	return time.Duration(ort.maxLatency.Load())
}

func (ort *observableRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
	start := time.Now()
	resp, err := ort.rt.RoundTrip(req)
	if err != nil {
		return resp, err
	}

	took := uint64(time.Since(start).Nanoseconds())
	for {
		max := ort.maxLatency.Load()
		if max >= took {
			return resp, err
		}
		if ort.maxLatency.CompareAndSwap(max, took) {
			return resp, err
		}
	}
}

Example (Instrumented)

package main

import (
	"net/http"
	"time"

	"github.com/cristalhq/hedgedhttp"
)

func main() {
	transport := &InstrumentedTransport{
		Transport: http.DefaultTransport,
	}

	_, err := hedgedhttp.NewRoundTripper(time.Millisecond, 3, transport)
	if err != nil {
		panic(err)
	}

}

type InstrumentedTransport struct {
	Transport http.RoundTripper
}

func (t *InstrumentedTransport) RoundTrip(req *http.Request) (*http.Response, error) {

	resp, err := t.Transport.RoundTrip(req)
	if err != nil {
		return nil, err
	}

	return resp, nil
}

Example (Ratelimited)

package main

import (
	"context"
	"errors"
	"math/rand"
	"net/http"
	"time"

	"github.com/cristalhq/hedgedhttp"
)

func main() {
	transport := &RateLimitedHedgedTransport{
		Transport: http.DefaultTransport,
		Limiter:   &RandomRateLimiter{},
	}

	_, err := hedgedhttp.NewRoundTripper(time.Millisecond, 3, transport)
	if err != nil {
		panic(err)
	}

}

// by example https://pkg.go.dev/golang.org/x/time/rate
type RateLimiter interface {
	Wait(ctx context.Context) error
}

type RateLimitedHedgedTransport struct {
	Transport http.RoundTripper
	Limiter   RateLimiter
}

func (t *RateLimitedHedgedTransport) RoundTrip(r *http.Request) (*http.Response, error) {

	if hedgedhttp.IsHedgedRequest(r) {
		if err := t.Limiter.Wait(r.Context()); err != nil {
			return nil, err
		}
	}
	return t.Transport.RoundTrip(r)
}

// Just for the example.
type RandomRateLimiter struct{}

func (r *RandomRateLimiter) Wait(ctx context.Context) error {
	if rand.Int()%2 == 0 {
		return errors.New("rate limit exceed")
	}
	return nil
}

Index

• func IsHedgedRequest(r *http.Request) bool
• func NewClient(timeout time.Duration, upto int, client *http.Client) (*http.Client, error)
• func NewRoundTripper(timeout time.Duration, upto int, rt http.RoundTripper) (http.RoundTripper, error)
• type Client
  • func New(cfg Config) (*Client, error)
  • func (c *Client) Do(req *http.Request) (*http.Response, error)
  • func (c *Client) RoundTrip(req *http.Request) (*http.Response, error)
  • func (c *Client) Stats() *Stats
• type Config
• type ErrorFormatFunc
• type MultiError
  • func (e *MultiError) Error() string
  • func (e *MultiError) ErrorOrNil() error
  • func (e *MultiError) String() string
• type NextFn
• type Stats
  • func NewClientAndStats(timeout time.Duration, upto int, client *http.Client) (*http.Client, *Stats, error)
  • func NewRoundTripperAndStats(timeout time.Duration, upto int, rt http.RoundTripper) (http.RoundTripper, *Stats, error)
  • func (s *Stats) ActualRoundTrips() uint64
  • func (s *Stats) CanceledByUserRoundTrips() uint64
  • func (s *Stats) CanceledSubRequests() uint64
  • func (s *Stats) FailedRoundTrips() uint64
  • func (s *Stats) HedgedRequestWins() uint64
  • func (s *Stats) OriginalRequestWins() uint64
  • func (s *Stats) RequestedRoundTrips() uint64
  • func (s *Stats) Snapshot() StatsSnapshot
• type StatsSnapshot

Examples

• Package (ConfigNext)
• Package (Instrumented)
• Package (Ratelimited)
• Client

Functions

func IsHedgedRequest

func IsHedgedRequest(r *http.Request) bool

IsHedgedRequest reports when a request is hedged.

func NewClient

func NewClient(timeout time.Duration, upto int, client *http.Client) (*http.Client, error)

NewClient returns a new http.Client which implements hedged requests pattern. Given Client starts a new request after a timeout from previous request. Starts no more than upto requests.

func NewRoundTripper

func NewRoundTripper(timeout time.Duration, upto int, rt http.RoundTripper) (http.RoundTripper, error)

NewRoundTripper returns a new http.RoundTripper which implements hedged requests pattern. Given RoundTripper starts a new request after a timeout from previous request. Starts no more than upto requests.

Types

type Client

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

Client represents a hedged HTTP client.

Example

package main

import (
	"context"
	"net/http"
	"time"

	"github.com/cristalhq/hedgedhttp"
)

func main() {
	ctx := context.Background()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://cristalhq.dev", http.NoBody)
	if err != nil {
		panic(err)
	}

	timeout := 10 * time.Millisecond
	upto := 7
	client := &http.Client{Timeout: time.Second}
	hedged, err := hedgedhttp.NewClient(timeout, upto, client)
	if err != nil {
		panic(err)
	}

	// will take `upto` requests, with a `timeout` delay between them
	resp, err := hedged.Do(req)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	// and do something with resp

}

func New

func New(cfg Config) (*Client, error)

New returns a new Client for the given config.

func (*Client) Do

func (c *Client) Do(req *http.Request) (*http.Response, error)

Do does the same as [RoundTrip], this method is presented to align with net/http.Client.

func (*Client) RoundTrip

func (c *Client) RoundTrip(req *http.Request) (*http.Response, error)

RoundTrip implements net/http.RoundTripper interface.

func (*Client) Stats

func (c *Client) Stats() *Stats

Stats returns statistics for the given client, see Stats methods.

type Config

type Config struct {
	// Transport of the [Client].
	// Default is nil which results in [net/http.DefaultTransport].
	Transport http.RoundTripper

	// Upto says how much requests to make.
	// Default is zero which means no hedged requests will be made.
	Upto int

	// Delay before 2 consequitive hedged requests.
	Delay time.Duration

	// Next returns the upto and delay for each HTTP that will be hedged.
	// Default is nil which results in (Upto, Delay) result.
	Next NextFn
}

Config for the Client.

type ErrorFormatFunc

type ErrorFormatFunc func([]error) string

ErrorFormatFunc is called by MultiError to return the list of errors as a string.

type MultiError

type MultiError struct {
	Errors        []error
	ErrorFormatFn ErrorFormatFunc
}

MultiError is an error type to track multiple errors. This is used to accumulate errors in cases and return them as a single "error". Inspired by https://github.com/hashicorp/go-multierror

func (*MultiError) Error

func (e *MultiError) Error() string

func (*MultiError) ErrorOrNil

func (e *MultiError) ErrorOrNil() error

ErrorOrNil returns an error if there are some.

func (*MultiError) String

func (e *MultiError) String() string

type NextFn

type NextFn func() (upto int, delay time.Duration)

NextFn represents a function that is called for each HTTP request for retrieving hedging options.

type Stats

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

Stats object that can be queried to obtain certain metrics and get better observability.

func NewClientAndStats

func NewClientAndStats(timeout time.Duration, upto int, client *http.Client) (*http.Client, *Stats, error)

NewClientAndStats returns a new http.Client which implements hedged requests pattern And Stats object that can be queried to obtain client's metrics. Given Client starts a new request after a timeout from previous request. Starts no more than upto requests.

func NewRoundTripperAndStats

func NewRoundTripperAndStats(timeout time.Duration, upto int, rt http.RoundTripper) (http.RoundTripper, *Stats, error)

NewRoundTripperAndStats returns a new http.RoundTripper which implements hedged requests pattern And Stats object that can be queried to obtain client's metrics. Given RoundTripper starts a new request after a timeout from previous request. Starts no more than upto requests.

func (*Stats) ActualRoundTrips

func (s *Stats) ActualRoundTrips() uint64

ActualRoundTrips returns count of requests that were actually sent.

func (*Stats) CanceledByUserRoundTrips

func (s *Stats) CanceledByUserRoundTrips() uint64

CanceledByUserRoundTrips returns count of requests that were canceled by user, using request context.

func (*Stats) CanceledSubRequests

func (s *Stats) CanceledSubRequests() uint64

CanceledSubRequests returns count of hedged sub-requests that were canceled by transport.

func (*Stats) FailedRoundTrips

func (s *Stats) FailedRoundTrips() uint64

FailedRoundTrips returns count of requests that failed.

func (*Stats) HedgedRequestWins

func (s *Stats) HedgedRequestWins() uint64

HedgedRequestWins returns count of hedged requests that were faster than the original.

func (*Stats) OriginalRequestWins

func (s *Stats) OriginalRequestWins() uint64

OriginalRequestWins returns count of original requests that were faster than the original.

func (*Stats) RequestedRoundTrips

func (s *Stats) RequestedRoundTrips() uint64

RequestedRoundTrips returns count of requests that were requested by client.

func (*Stats) Snapshot

func (s *Stats) Snapshot() StatsSnapshot

Snapshot of the stats.

type StatsSnapshot

type StatsSnapshot struct {
	RequestedRoundTrips      uint64 // count of requests that were requested by client
	ActualRoundTrips         uint64 // count of requests that were actually sent
	FailedRoundTrips         uint64 // count of requests that failed
	CanceledByUserRoundTrips uint64 // count of requests that were canceled by user, using request context
	CanceledSubRequests      uint64 // count of hedged sub-requests that were canceled by transport
}

StatsSnapshot is a snapshot of Stats.