apps/streamctl
1
0
Fork 0
mirror of https://github.com/xaionaro-go/streamctl.git synced 2025-10-12 10:50:05 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Diagnoze why memoization does not work

Browse Source
This commit is contained in:
Dmitrii Okunev
2025-06-29 20:22:02 +01:00
parent 32e6aec5cd
commit 2d9ae83ec3
2 changed files with 11 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pkg/streamd/memoize/memoize.go
View File

@@ -39,6 +39,7 @@ func Memoize[REQ any, REPLY any, T func(context.Context, REQ) (REPLY, error)](
defer logger.Tracef(ctx, "/memoize %T:%#+v", req, req) defer logger.Tracef(ctx, "/memoize %T:%#+v", req, req)
if IsNoCache(ctx) { if IsNoCache(ctx) {
logger.Tracef(ctx, "forcing no cache")
cacheDuration = 0 cacheDuration = 0
} }
@@ -48,6 +49,7 @@ func Memoize[REQ any, REPLY any, T func(context.Context, REQ) (REPLY, error)](
return fn(ctx, req) return fn(ctx, req)
} }
ctx = belt.WithField(ctx, "cache_key", string(key[:])) ctx = belt.WithField(ctx, "cache_key", string(key[:]))
ctx = belt.WithField(ctx, "cache_dur", cacheDuration)
logger.Tracef(ctx, "req: %T:%#+v", req, req) logger.Tracef(ctx, "req: %T:%#+v", req, req)
d.CacheMetaLock.Lock() d.CacheMetaLock.Lock()
@@ -80,7 +82,8 @@ func Memoize[REQ any, REPLY any, T func(context.Context, REQ) (REPLY, error)](
if ok { if ok {
if v, ok := cachedResult.(cacheItem); ok { if v, ok := cachedResult.(cacheItem); ok {
cutoffTS := time.Now().Add(-cacheDuration) now := time.Now()
cutoffTS := now.Add(-cacheDuration)
if cacheDuration > 0 && !v.SavedAt.Before(cutoffTS) { if cacheDuration > 0 && !v.SavedAt.Before(cutoffTS) {
d.CacheMetaLock.Unlock() d.CacheMetaLock.Unlock()
logger.Tracef(ctx, "grpc.CacheMetaLock.Unlock()-ed") logger.Tracef(ctx, "grpc.CacheMetaLock.Unlock()-ed")
@@ -92,9 +95,10 @@ func Memoize[REQ any, REPLY any, T func(context.Context, REQ) (REPLY, error)](
} }
logger.Tracef( logger.Tracef(
ctx, ctx,
"the cached value expired: %s < %s", "the cached value expired: %s < %s (%s - %s)",
v.SavedAt.Format(timeFormat), v.SavedAt.Format(timeFormat),
cutoffTS.Format(timeFormat), cutoffTS.Format(timeFormat),
now.Format(timeFormat), cacheDuration,
) )
delete(cache, key) delete(cache, key)
} else { } else {

6
pkg/streamd/streamd.go
View File

@@ -937,10 +937,14 @@ func (d *StreamD) streamController(
} }
return result, nil return result, nil
} }
func (d *StreamD) GetStreamStatus( func (d *StreamD) GetStreamStatus(
ctx context.Context, ctx context.Context,
platID streamcontrol.PlatformName, platID streamcontrol.PlatformName,
) (*streamcontrol.StreamStatus, error) { ) (_ret *streamcontrol.StreamStatus, _err error) {
ctx = belt.WithField(ctx, "plat_id", platID)
logger.Tracef(ctx, "GetStreamStatus(ctx, '%s')", platID)
defer func() { logger.Tracef(ctx, "/GetStreamStatus(ctx, '%s'): %#+v %v", platID, _ret, _err) }()
cacheDuration := 5 * time.Second cacheDuration := 5 * time.Second
switch platID { switch platID {
case obs.ID: case obs.ID: