حول التدافع مخبأ

غالبًا ما ينتهي بي الأمر في المواقف التي أحتاج فيها إلى تخزين هذا أو ذاك مؤقتًا. في كثير من الأحيان، يتم تخزين هذه القيم مؤقتًا لفترة من الوقت. ربما تكون على دراية بالنمط. تحاول الحصول على قيمة من ذاكرة التخزين المؤقت، إذا نجحت، تقوم بإعادتها إلى المتصل واستدعاءها يوميًا. إذا لم تكن القيمة موجودة، يمكنك إحضارها (على الأرجح من قاعدة البيانات) أو حسابها ووضعها في ذاكرة التخزين المؤقت. في معظم الحالات، يعمل هذا بشكل رائع. ومع ذلك، إذا تم الوصول إلى المفتاح الذي تستخدمه لإدخال ذاكرة التخزين المؤقت بشكل متكرر واستغرقت عملية حساب البيانات بعض الوقت، فسوف ينتهي بك الأمر في موقف حيث ستفقد الطلبات المتوازية المتعددة في وقت واحد ذاكرة التخزين المؤقت. ستقوم جميع هذه الطلبات بتحميل المصدر بشكل مستقل وتخزين القيمة في ذاكرة التخزين المؤقت. يؤدي هذا إلى إهدار الموارد ويمكن أن يؤدي أيضًا إلى رفض الخدمة.

اسمحوا لي أن أوضح بمثال. سأستخدم redis لذاكرة التخزين المؤقت وخادم Go http البسيط في الأعلى. إليك الكود الكامل:

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())
    }
}

فلنضع بعض التحميل على نقطة النهاية /simple ونرى ما سيحدث. سأستخدم فيجيتا لهذا.

أجري هجوم فيجيتا -duration=30s -rate=500 -targets=./targets_simple.txt > res_simple.bin. ينتهي الأمر بـ Vegeta بتقديم 500 طلب كل ثانية لمدة 30 ثانية. أقوم برسمها بيانيًا على شكل رسم بياني لرموز نتائج HTTP مع مجموعات يبلغ طول كل منها 100 مللي ثانية. والنتيجة هي الرسم البياني التالي.

عندما نبدأ التجربة، تكون ذاكرة التخزين المؤقت فارغة - وليس لدينا أي قيمة مخزنة هناك. نحصل على التدافع الأولي عندما تصل مجموعة من الطلبات إلى خادمنا. كلهم يقومون بفحص ذاكرة التخزين المؤقت، ولم يجدوا أي شيء هناك، واستدعاء عملية longRunningOperation وتخزينها في ذاكرة التخزين المؤقت. نظرًا لأن عملية longRunningOperation تستغرق حوالي 500 مللي ثانية لإكمال أي طلبات تم تقديمها في أول 500 مللي ثانية، ينتهي الأمر باستدعاء عملية longRunningOperation. بمجرد أن يتمكن أحد الطلبات من تخزين القيمة في ذاكرة التخزين المؤقت، تقوم جميع الطلبات التالية بإحضارها من ذاكرة التخزين المؤقت ونبدأ في رؤية الاستجابات برمز الحالة 200. ثم يتكرر النمط كل 3 ثوانٍ مع بدء تشغيل آلية انتهاء الصلاحية على redis.

في مثال اللعبة هذا، لا يسبب هذا أي مشكلات ولكن في بيئة الإنتاج، يمكن أن يؤدي ذلك إلى تحميل غير ضروري على أنظمتك، أو تدهور تجربة المستخدم أو حتى رفض الخدمة ذاتيًا. فكيف يمكننا منع هذا؟ حسنًا، هناك عدة طرق. يمكننا تقديم قفل - أي خطأ في ذاكرة التخزين المؤقت قد يؤدي إلى محاولة الكود تحقيق القفل. لا يعد القفل الموزع أمرًا تافهًا، وغالبًا ما تكون هذه الحالات ذات حواف دقيقة تتطلب معالجة دقيقة. يمكننا أيضًا إعادة حساب القيمة بشكل دوري باستخدام وظيفة في الخلفية، لكن هذا يتطلب عملية إضافية ليتم تشغيلها، مما يوفر ترسًا آخر يحتاج إلى الصيانة والمراقبة في التعليمات البرمجية الخاصة بنا. قد لا يكون هذا الأسلوب ممكنًا أيضًا إذا كان لديك مفاتيح ذاكرة تخزين مؤقت ديناميكية. هناك نهج آخر يسمى انتهاء الصلاحية المبكر الاحتمالي وهذا شيء أود استكشافه بشكل أكبر.

احتمال انتهاء الصلاحية المبكر

تسمح هذه التقنية بإعادة حساب القيمة بناءً على الاحتمال. عند جلب القيمة من ذاكرة التخزين المؤقت، يمكنك أيضًا حساب ما إذا كنت بحاجة إلى إعادة إنشاء قيمة ذاكرة التخزين المؤقت بناءً على الاحتمال. كلما اقتربت من انتهاء القيمة الحالية، كلما زاد الاحتمال.

أعتمد التنفيذ المحدد على XFetch بواسطة A. Vattani وF.Chierichetti & K. Lowenstein في منع التدافع الاحتمالي الأمثل لذاكرة التخزين المؤقت.

سأقدم نقطة نهاية جديدة على خادم HTTP والتي ستقوم أيضًا بإجراء عمليات حسابية باهظة الثمن ولكن هذه المرة استخدم XFetch عند التخزين المؤقت. لكي يعمل XFetch، نحتاج إلى تخزين المدة التي استغرقتها العملية الباهظة الثمن (الدلتا) ومتى تنتهي صلاحية مفتاح ذاكرة التخزين المؤقت. ولتحقيق ذلك، سأقدم بنية تحمل هذه القيم بالإضافة إلى الرسالة نفسها:

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

أضيف وظيفة لتغليف الرسالة الأصلية بهذه السمات وإجراء تسلسل لها لتخزينها في 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
}

لنكتب أيضًا طريقة لإعادة حساب القيمة وتخزينها في 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
}

لتحديد ما إذا كنا بحاجة إلى تحديث القيمة بناءً على الاحتمالية، يمكننا إضافة طريقة إلى القيمة الاحتمالية:

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
}

إذا قمنا بربط كل شيء، سننتهي بالمعالج التالي:

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

يعمل المعالج بشكل يشبه إلى حد كبير المعالج الأول، ومع ذلك، عند الحصول على ذاكرة تخزين مؤقت، نقوم برمي النرد. اعتمادًا على النتيجة، إما أن نعيد القيمة التي جلبناها للتو، أو نقوم بتحديث القيمة مبكرًا.

سنستخدم رموز حالة HTTP لتحديد بين الحالات الثلاث:

• 200 - قمنا بإرجاع القيمة من ذاكرة التخزين المؤقت
• 201 - ذاكرة التخزين المؤقت مفقودة، لا توجد قيمة موجودة
• 202 - تم الوصول إلى ذاكرة التخزين المؤقت، مما أدى إلى التحديث الاحتمالي

لقد قمت بتشغيل vegeta مرة أخرى هذه المرة مقابل نقطة النهاية الجديدة وهذه هي النتيجة:

تشير النقط الزرقاء الصغيرة هناك إلى الوقت الذي انتهينا فيه بالفعل من تحديث قيمة ذاكرة التخزين المؤقت مبكرًا. لم نعد نرى أي فقدان لذاكرة التخزين المؤقت بعد فترة الإحماء الأولية. لتجنب الارتفاع الأولي، يمكنك تخزين القيمة المخزنة مؤقتًا مسبقًا إذا كان ذلك مهمًا لحالة الاستخدام الخاصة بك.

إذا كنت ترغب في أن تكون أكثر جرأة في التخزين المؤقت وتحديث القيمة بشكل متكرر، فيمكنك اللعب باستخدام المعلمة التجريبية. إليك ما تبدو عليه نفس التجربة مع تعيين المعلمة التجريبية على 2:

إننا نشهد الآن تحديثات احتمالية بشكل متكرر أكثر.

الكل في الكل يعد هذا أسلوبًا صغيرًا أنيقًا يمكن أن يساعد في تجنب التدافع في ذاكرة التخزين المؤقت. ومع ذلك، ضع في اعتبارك أن هذا لا يعمل إلا إذا كنت تقوم بشكل دوري بجلب نفس المفتاح من ذاكرة التخزين المؤقت - وإلا فلن ترى فائدة كبيرة.

هل لديك طريقة أخرى للتعامل مع حالات التدافع في ذاكرة التخزين المؤقت؟ هل لاحظت خطأ؟ اسمحوا لي أن أعرف في التعليقات أدناه!