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 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

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 consecutive 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
	// contains filtered or unexported fields
}

Config for the Client.

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.