Expiration anticipée probabiliste en Go-Golang-php.cn

À propos des bousculades de caches

Je me retrouve souvent dans des situations où j'ai besoin de mettre en cache ceci ou cela. Souvent, ces valeurs sont mises en cache pendant un certain temps. Vous connaissez probablement le modèle. Vous essayez d'obtenir une valeur du cache, si vous réussissez, vous la renvoyez à l'appelant et l'appelez un jour. Si la valeur n'est pas là, vous la récupérez (très probablement à partir de la base de données) ou la calculez et la mettez dans le cache. Dans la plupart des cas, cela fonctionne très bien. Cependant, si la clé que vous utilisez pour votre entrée de cache est fréquemment consultée et que l'opération de calcul des données prend un certain temps, vous vous retrouverez dans une situation où plusieurs requêtes parallèles obtiendront simultanément un échec de cache. Toutes ces requêtes chargeront indépendamment la source et stockeront la valeur dans le cache. Cela entraîne un gaspillage de ressources et peut même conduire à un déni de service.

Permettez-moi d'illustrer avec un exemple. J'utiliserai Redis pour le cache et un simple serveur Go http en plus. Voici le code complet :

package main

import (
    "errors"
    "log"
    "net/http"
    "time"

    "github.com/redis/go-redis/v9"
)

type handler struct {
    rdb *redis.Client
    cacheTTL time.Duration
}

func (ch *handler) simple(w http.ResponseWriter, r *http.Request) {
    cacheKey := "my_cache_key"
    // we'll use 200 to signify a cache hit & 201 to signify a miss
    responseCode := http.StatusOK
    cachedData, err := ch.rdb.Get(r.Context(), cacheKey).Result()
    if err != nil {
        if !errors.Is(err, redis.Nil) {
            log.Println("could not reach redis", err.Error())
            http.Error(w, "could not reach redis", http.StatusInternalServerError)
            return
        }

        // cache miss - fetch & store
        res := longRunningOperation()
        responseCode = http.StatusCreated

        err = ch.rdb.Set(r.Context(), cacheKey, res, ch.cacheTTL).Err()
        if err != nil {
            log.Println("failed to set cache value", err.Error())
            http.Error(w, "failed to set cache value", http.StatusInternalServerError)
            return
        }
        cachedData = res
    }
    w.WriteHeader(responseCode)
    _, _ = w.Write([]byte(cachedData))
}

func longRunningOperation() string {
    time.Sleep(time.Millisecond * 500)
    return "hello"
}

func main() {
    ttl := time.Second * 3
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })

    handler := &handler{
        rdb: rdb,
        cacheTTL: ttl,
    }

    http.HandleFunc("/simple", handler.simple)
    if err := http.ListenAndServe(":8080", nil); err != nil {
        log.Fatalf("Could not start server: %s\n", err.Error())
    }
}

Copier après la connexion

Mettons un peu de charge sur le point de terminaison /simple et voyons ce qui se passe. J'utiliserai Vegeta pour ça.

Je lance une attaque Vegeta -duration=30s -rate=500 -targets=./targets_simple.txt > res_simple.bin. Vegeta finit par faire 500 requêtes chaque seconde pendant 30 secondes. Je les représente sous forme d'histogramme de codes de résultat HTTP avec des compartiments d'une durée de 100 ms chacun. Le résultat est le graphique suivant.

Probabilistic Early Expiration in Go

Lorsque nous démarrons l'expérience, le cache est vide - nous n'y avons aucune valeur stockée. Nous obtenons la bousculade initiale alors qu’un tas de demandes atteignent notre serveur. Tous vérifient le cache et n'y trouvent rien, appellent longRunningOperation et le stockent dans le cache. Étant donné que longRunningOperation prend environ 500 ms pour terminer, toutes les requêtes effectuées au cours des 500 premières ms finissent par appeler longRunningOperation. Une fois que l'une des requêtes parvient à stocker la valeur dans le cache, toutes les requêtes suivantes la récupèrent du cache et nous commençons à voir des réponses avec le code d'état de 200. Le modèle se répète ensuite toutes les 3 secondes lorsque le mécanisme d'expiration sur Redis entre en jeu.

Dans cet exemple de jouet, cela ne pose aucun problème, mais dans un environnement de production, cela peut entraîner une charge inutile sur vos systèmes, une expérience utilisateur dégradée ou même un déni de service auto-induit. Alors, comment pouvons-nous empêcher cela ? Eh bien, il existe plusieurs façons. Nous pourrions introduire un verrou - tout échec de cache entraînerait une tentative de code d'obtenir un verrou. Le verrouillage distribué n'est pas une chose triviale à faire et il présente souvent des cas extrêmes subtils qui nécessitent une manipulation délicate. Nous pourrions également recalculer périodiquement la valeur à l'aide d'un travail en arrière-plan, mais cela nécessite l'exécution d'un processus supplémentaire, introduisant encore un autre rouage qui doit être maintenu et surveillé dans notre code. Cette approche peut également ne pas être réalisable si vous disposez de clés de cache dynamiques. Il existe une autre approche, appelée expiration anticipée probabiliste, et c'est quelque chose que j'aimerais approfondir.

Expiration anticipée probabiliste

Cette technique permet de recalculer la valeur en fonction d'une probabilité. Lors de la récupération de la valeur du cache, vous calculez également si vous devez régénérer la valeur du cache en fonction d'une probabilité. Plus vous êtes proche de l'expiration de la valeur existante, plus la probabilité est élevée.

Je base l'implémentation spécifique sur XFetch par A. Vattani, F.Chierichetti et K. Lowenstein dans Optimal Probabilistic Cache Stampede Prevention.

Je vais introduire un nouveau point de terminaison sur le serveur HTTP qui effectuera également le calcul coûteux mais cette fois utilisera XFetch lors de la mise en cache. Pour que XFetch fonctionne, nous devons stocker la durée de l'opération coûteuse (le delta) et la date d'expiration de la clé de cache. Pour y parvenir, je vais introduire une structure qui contiendra ces valeurs ainsi que le message lui-même :

type probabilisticValue struct {
    Message string
    Expiry time.Time
    Delta time.Duration
}

Copier après la connexion

J'ajoute une fonction pour envelopper le message d'origine avec ces attributs et le sérialiser pour le stocker dans Redis :

func wrapMessage(message string, delta, cacheTTL time.Duration) (string, error) {
    bts, err := json.Marshal(probabilisticValue{
        Message: message,
        Delta: delta,
        Expiry: time.Now().Add(cacheTTL),
    })
    if err != nil {
        return "", fmt.Errorf("could not marshal message: %w", err)
    }

    return string(bts), nil
}

Copier après la connexion

Écrivons également une méthode pour recalculer et stocker la valeur dans redis :

func (ch *handler) recomputeValue(ctx context.Context, cacheKey string) (string, error) {
    start := time.Now()
    message := longRunningOperation()
    delta := time.Since(start)

    wrapped, err := wrapMessage(message, delta, ch.cacheTTL)
    if err != nil {
        return "", fmt.Errorf("could not wrap message: %w", err)
    }
    err = ch.rdb.Set(ctx, cacheKey, wrapped, ch.cacheTTL).Err()
    if err != nil {
        return "", fmt.Errorf("could not save value: %w", err)
    }
    return message, nil
}

Copier après la connexion

Pour déterminer si nous devons mettre à jour la valeur en fonction de la probabilité, nous pouvons ajouter une méthode à probabilisticValue :

func (pv probabilisticValue) shouldUpdate() bool {
    // suggested default param in XFetch implementation
    // if increased - results in earlier expirations
    beta := 1.0
    now := time.Now()
    scaledGap := pv.Delta.Seconds() * beta * math.Log(rand.Float64())
    return now.Sub(pv.Expiry).Seconds() >= scaledGap
}

Copier après la connexion

Si nous connectons tout cela, nous nous retrouvons avec le gestionnaire suivant :

func (ch *handler) probabilistic(w http.ResponseWriter, r *http.Request) {
    cacheKey := "probabilistic_cache_key"
    // we'll use 200 to signify a cache hit & 201 to signify a miss
    responseCode := http.StatusOK
    cachedData, err := ch.rdb.Get(r.Context(), cacheKey).Result()
    if err != nil {
        if !errors.Is(err, redis.Nil) {
            log.Println("could not reach redis", err.Error())
            http.Error(w, "could not reach redis", http.StatusInternalServerError)
            return
        }

        res, err := ch.recomputeValue(r.Context(), cacheKey)
        if err != nil {
            log.Println("could not recompute value", err.Error())
            http.Error(w, "could not recompute value", http.StatusInternalServerError)
            return
        }
        responseCode = http.StatusCreated
        cachedData = res

        w.WriteHeader(responseCode)
        _, _ = w.Write([]byte(cachedData))
        return
    }

    pv := probabilisticValue{}
    err = json.Unmarshal([]byte(cachedData), &pv)
    if err != nil {
        log.Println("could not unmarshal probabilistic value", err.Error())
        http.Error(w, "could not unmarshal probabilistic value", http.StatusInternalServerError)
        return
    }

    if pv.shouldUpdate() {
        _, err := ch.recomputeValue(r.Context(), cacheKey)
        if err != nil {
            log.Println("could not recompute value", err.Error())
            http.Error(w, "could not recompute value", http.StatusInternalServerError)
            return
        }
        responseCode = http.StatusAccepted
    }

    w.WriteHeader(responseCode)
    _, _ = w.Write([]byte(cachedData))
}

Copier après la connexion

Le gestionnaire fonctionne un peu comme le premier, cependant, dès que nous obtenons un accès au cache, nous lançons les dés. En fonction du résultat, soit nous renvoyons simplement la valeur que nous venons de récupérer, soit nous mettons à jour la valeur plus tôt.

Nous utiliserons les codes de statut HTTP pour déterminer entre les 3 cas :