In a golang backend I want to serve a value to a multiple clients, lets call it score. Score is changing with time, and its calculation is slow. Calculation does not depend on previous results. When there are no clients I dont want to calculate it at all. So calculation should happen only on request. But also there is another fact - score cannot change within 5 seconds period. So i tried different aproaches and everything has its drawbacks:

1. Does expensive calculation in absense of clients:

var score interface{}

// run in a separate goroutine
func calculateScorePeriodically() {
    for{
        select{
        case <-time.After(5*time.Second):
            score = calculateScoreExpensiveAndSlow()
        }
    }
}

func serveScore(w http.ResponseWriter, r* http.Request) {
    b, _ := json.Marshal(score)
    w.Write(b)
}

2. Blocks all clients for a long calculation period (but actually may just serve old data to them). And you cannot move if outside a mutex, because then multiple client may enter calculation block simultaneously and would do calculation not within 5 seconds interval but sequentially:

var (
    score interface{}
    mutex sync.Mutex
    updatedAt time.Time
)

func getCachedScore() float64 {
    mutex.Lock()
    defer mutex.Unlock()
    currentTime := time.Now()
    if currentTime.Sub(updatedAt) < 5*time.Second {
        return score
    }
    updatedAt = currentTime
    score = calculateScoreExpensiveAndSlow()
    return score
}

func serveScore(w http.ResponseWriter, r* http.Request) {
    b, _ := json.Marshal(getCachedScore())
    w.Write(b)
}

How to solve both of above drawbacks?

PS. i think this is a generic problem, and a pattern - does it have a special name?

CodePudding user response：

There may be multiple solutions. A simple solution is to have a designated goroutine for calculation, to which you can signal a need for recalculation by sending a value on a channel. The send may be non-blocking, so if a calculation is in progress, nothing will happen.

Here's a reusable cache implementation:

type cache struct {
    mu      sync.RWMutex
    value   interface{}
    updated time.Time

    calcCh     chan struct{}
    expiration time.Duration
}

func NewCache(calc func() interface{}, expiration time.Duration) *cache {
    c := &cache{
        value:   calc(),
        updated: time.Now(),
        calcCh:  make(chan struct{}),
    }

    go func() {
        for range c.calcCh {
            v := calc()

            c.mu.Lock()
            c.value, c.updated = v, time.Now()
            c.mu.Unlock()
        }
    }()

    return c
}

func (c *cache) Get() (value interface{}, updated time.Time) {
    c.mu.RLock()
    value, updated = c.value, c.updated
    c.mu.RUnlock()

    if time.Since(updated) > c.expiration {
        // Trigger a new calculation (will happen in another goroutine).
        // Do non-blocking send, if a calculation is in progress,
        // this will have no effect
        select {
        case c.calcCh <- struct{}{}:
        default:
        }
    }

    return
}

func (c *cache) Stop() {
    close(c.calcCh)
}

Note: Cache.Stop() is to stop the background goroutine. After calling Cache.Stop(), Cache.Get() must not be called.

Using it for your case:

func getCachedScore() interface{} {
    // ...
}

var scoreCache = NewCache(getCachedScore, 5*time.Second)

func serveScore(w http.ResponseWriter, r* http.Request) {
    score, _ := scoreCache.Get()
    b, _ := json.Marshal(score)
    w.Write(b)
}

CodePudding user response：

This is what i've implemented, correlates with icza's answer, but has some more features:

package common

import (
    "context"
    "sync"
    "sync/atomic"
    "time"
)

type (
    SyncValueUpdateFunc func() interface{}
    ChanStruct          chan struct{}
)

type SyncValue struct {
    value        atomic.Value  // holds the value interface{}
    updatedAt    atomic.Value  // holds time.Time, time when last update sequence was started at
    updatePeriod time.Duration // controls minimal anount of time between updates
    needUpdate   ChanStruct
}

// updater is a user-provided function with long expensive calculation, that gets current state
func MakeSyncValue(ctx context.Context, updatePeriod time.Duration, updater SyncValueUpdateFunc) *SyncValue {
    v := &SyncValue{
        updatePeriod: updatePeriod,
        needUpdate:   make(ChanStruct, 256),
    }
    v.updatedAt.Store(time.Now()) // "was never updated", but time should never be nil interface
    v.value.Store(updater())
    go v.updateController(ctx, updater)
    return v
}

//client will get value immediately, and optionally may trigger an update. may return nil in first calls (threat it as "cache was not updated yet")
func (v *SyncValue) Get() interface{} {
    if v.IsExpired(time.Now()) {
        v.RequestUpdate()
    }
    return v.value.Load()
}

//updateController goroutine can be terminated both by context or by closing channel
func (v *SyncValue) Stop() {
    close(v.needUpdate)
}

//returns true if value is outdated and updater function is not called yet
func (v *SyncValue) IsExpired(currentTime time.Time) bool {
    updatedAt := v.updatedAt.Load().(time.Time)
    return currentTime.Sub(updatedAt) > v.updatePeriod
}

// updateController can decide not to update cached value if controller has updated value recently
func (v *SyncValue) RequestUpdate() {
    v.needUpdate <- struct{}{}
}

func (v *SyncValue) updateController(ctx context.Context, updater SyncValueUpdateFunc) {
    var wg sync.WaitGroup
    wg.Wait()
    for {
        select {
        case <-ctx.Done():
            return
        case _, ok := <-v.needUpdate:
            if !ok {
                return
            }
            currentTime := time.Now()
            if !v.IsExpired(currentTime) {
                continue
            }
            v.updatedAt.Store(currentTime) // store it first, so clients would likely not request updates, while update is in process
            v.value.Store(updater())
        }
    }
}