⛅ Sync 2026-08-22 23:43:25

mosdns/patches/213-feat-cache-add-proactive-prefetch-support.patch

@@ -0,0 +1,227 @@
+From 819415303eb05bdf5b6dadf119f947645c9062cc Mon Sep 17 00:00:00 2001
+From: sbwml <admin@cooluc.com>
+Date: Sat, 22 Aug 2026 22:56:30 +0800
+Subject: [PATCH] feat(cache): add proactive prefetch support
+MIME-Version: 1.0
+Content-Type: text/plain; charset=UTF-8
+Content-Transfer-Encoding: 8bit
+
+plugins:
+  - tag: lazy_cache
+    type: cache
+    args:
+      size: 1024
+      # 开启预读功能 (默认为 false)
+      prefetch: true
+      # 在 TTL 还剩 10 秒时尝试预读 (默认为 10)
+      prefetch_before_expire: 10
+      # 至少被命中 2 次才触发预读 (默认为 3)
+      prefetch_min_hits: 2
+      # 后台扫描周期，每 3 秒扫描一次 (默认为 5)
+      prefetch_scan_interval: 3
+
+Signed-off-by: sbwml <admin@cooluc.com>
+---
+ plugin/executable/cache/cache.go | 113 +++++++++++++++++++++++++++++--
+ plugin/executable/cache/utils.go |   2 +
+ 2 files changed, 110 insertions(+), 5 deletions(-)
+
+--- a/plugin/executable/cache/cache.go
++++ b/plugin/executable/cache/cache.go
+@@ -27,6 +27,7 @@ import (
+ 	"io"
+ 	"net/http"
+ 	"os"
++	"reflect"
+ 	"strconv"
+ 	"sync"
+ 	"sync/atomic"
+@@ -69,15 +70,22 @@ const (
+ var _ sequence.RecursiveExecutable = (*Cache)(nil)
+ 
+ type Args struct {
+-	Size         int    `yaml:"size"`
+-	LazyCacheTTL int    `yaml:"lazy_cache_ttl"`
+-	DumpFile     string `yaml:"dump_file"`
+-	DumpInterval int    `yaml:"dump_interval"`
++	Size                 int    `yaml:"size"`
++	LazyCacheTTL         int    `yaml:"lazy_cache_ttl"`
++	DumpFile             string `yaml:"dump_file"`
++	DumpInterval         int    `yaml:"dump_interval"`
++	Prefetch             bool   `yaml:"prefetch"`
++	PrefetchBeforeExpire int    `yaml:"prefetch_before_expire"`
++	PrefetchMinHits      int    `yaml:"prefetch_min_hits"`
++	PrefetchScanInterval int    `yaml:"prefetch_scan_interval"`
+ }
+ 
+ func (a *Args) init() {
+ 	utils.SetDefaultUnsignNum(&a.Size, 1024)
+ 	utils.SetDefaultUnsignNum(&a.DumpInterval, 600)
++	utils.SetDefaultUnsignNum(&a.PrefetchBeforeExpire, 10)
++	utils.SetDefaultUnsignNum(&a.PrefetchMinHits, 3)
++	utils.SetDefaultUnsignNum(&a.PrefetchScanInterval, 5)
+ }
+ 
+ type Cache struct {
+@@ -94,6 +102,11 @@ type Cache struct {
+ 	hitTotal     prometheus.Counter
+ 	lazyHitTotal prometheus.Counter
+ 	size         prometheus.GaugeFunc
++
++	prefetchTotal     prometheus.Counter
++	prefetchFailTotal prometheus.Counter
++	capturedNext      *sequence.ChainWalker
++	captureOnce       sync.Once
+ }
+ 
+ func Init(bp *coremain.BP, args any) (any, error) {
+@@ -160,6 +173,16 @@ func NewCache(args *Args, opts Opts) *Ca
+ 			Help:        "The total number of queries that hit the expired cache",
+ 			ConstLabels: lb,
+ 		}),
++		prefetchTotal: prometheus.NewCounter(prometheus.CounterOpts{
++			Name:        "prefetch_total",
++			Help:        "The total number of successful proactive prefetch operations",
++			ConstLabels: lb,
++		}),
++		prefetchFailTotal: prometheus.NewCounter(prometheus.CounterOpts{
++			Name:        "prefetch_fail_total",
++			Help:        "The total number of failed proactive prefetch operations",
++			ConstLabels: lb,
++		}),
+ 		size: prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+ 			Name:        "size_current",
+ 			Help:        "Current cache size in records",
+@@ -173,12 +196,15 @@ func NewCache(args *Args, opts Opts) *Ca
+ 		p.logger.Error("failed to load cache dump", zap.Error(err))
+ 	}
+ 	p.startDumpLoop()
++	if p.args.Prefetch {
++		p.startPrefetchLoop()
++	}
+ 
+ 	return p
+ }
+ 
+ func (c *Cache) RegMetricsTo(r prometheus.Registerer) error {
+-	for _, collector := range [...]prometheus.Collector{c.queryTotal, c.hitTotal, c.lazyHitTotal, c.size} {
++	for _, collector := range [...]prometheus.Collector{c.queryTotal, c.hitTotal, c.lazyHitTotal, c.prefetchTotal, c.prefetchFailTotal, c.size} {
+ 		if err := r.Register(collector); err != nil {
+ 			return err
+ 		}
+@@ -187,6 +213,15 @@ func (c *Cache) RegMetricsTo(r prometheu
+ }
+ 
+ func (c *Cache) Exec(ctx context.Context, qCtx *query_context.Context, next sequence.ChainWalker) error {
++	c.captureOnce.Do(func() {
++		// This implementation assumes the cache plugin is invoked from a single fixed position in the sequence.
++		nextCopy := next
++		c.capturedNext = &nextCopy
++	})
++	if c.capturedNext != nil && !reflect.DeepEqual(next, *c.capturedNext) {
++		c.logger.Warn("cache plugin Exec called with a different next chain than first time")
++	}
++
+ 	c.queryTotal.Inc()
+ 	q := qCtx.Q()
+ 
+@@ -205,6 +240,7 @@ func (c *Cache) Exec(ctx context.Context
+ 		cachedResp.Id = q.Id // change msg id
+ 		qCtx.SetResponse(cachedResp)
+ 		if v, _, ok := c.backend.Get(key(msgKey)); ok && v != nil {
++			v.hitCount.Add(1)
+ 			ttl := int(v.expirationTime.Sub(v.storedTime).Seconds())
+ 			remainingTtl := int(v.expirationTime.Sub(time.Now()).Seconds())
+ 			if remainingTtl < 0 {
+@@ -255,6 +291,73 @@ func (c *Cache) doLazyUpdate(msgKey stri
+ 	c.lazyUpdateSF.DoChan(msgKey, lazyUpdateFunc) // DoChan won't block this goroutine
+ }
+ 
++func (c *Cache) startPrefetchLoop() {
++	go func() {
++		ticker := time.NewTicker(time.Duration(c.args.PrefetchScanInterval) * time.Second)
++		defer ticker.Stop()
++		for {
++			select {
++			case <-ticker.C:
++				if c.capturedNext == nil {
++					continue
++				}
++				now := time.Now()
++				_ = c.backend.Range(func(k key, v *item, cacheExpirationTime time.Time) error {
++					remainingTTL := int(v.expirationTime.Sub(now).Seconds())
++					if remainingTTL >= 0 && remainingTTL < c.args.PrefetchBeforeExpire && v.hitCount.Load() >= uint32(c.args.PrefetchMinHits) {
++						c.doPrefetch(string(k), v, remainingTTL)
++					}
++					return nil
++				})
++			case <-c.closeNotify:
++				return
++			}
++		}
++	}()
++}
++
++func (c *Cache) doPrefetch(msgKey string, v *item, remainingTTL int) {
++	prefetchFunc := func() (any, error) {
++		defer c.lazyUpdateSF.Forget(msgKey)
++
++		req := new(dns.Msg)
++		req.SetQuestion(v.resp.Question[0].Name, v.resp.Question[0].Qtype)
++		req.Id = dns.Id()
++		req.RecursionDesired = true
++		newQCtx := query_context.NewContext(req)
++
++		qname := v.resp.Question[0].Name
++		hitCount := v.hitCount.Load()
++
++		c.logger.Debug("start proactive prefetch",
++			zap.String("qname", qname),
++			zap.Uint32("hit_count", hitCount),
++			zap.Int("remaining_ttl", remainingTTL),
++		)
++		ctx, cancel := context.WithTimeout(context.Background(), defaultLazyUpdateTimeout)
++		defer cancel()
++
++		err := c.capturedNext.ExecNext(ctx, newQCtx)
++		if err != nil {
++			c.logger.Warn("failed to proactive prefetch", zap.String("qname", qname), zap.Error(err))
++			c.prefetchFailTotal.Inc()
++			return nil, err
++		}
++
++		r := newQCtx.R()
++		if r != nil {
++			if saveRespToCache(msgKey, r, c.backend, c.args.LazyCacheTTL) {
++				v.hitCount.Store(0)
++				c.prefetchTotal.Inc()
++				c.updatedKey.Add(1)
++			}
++		}
++		c.logger.Debug("proactive prefetch updated", zap.String("qname", qname), newQCtx.InfoField())
++		return nil, nil
++	}
++	c.lazyUpdateSF.DoChan(msgKey, prefetchFunc)
++}
++
+ func (c *Cache) Close() error {
+ 	if err := c.dumpCache(); err != nil {
+ 		c.logger.Error("failed to dump cache", zap.Error(err))
+--- a/plugin/executable/cache/utils.go
++++ b/plugin/executable/cache/utils.go
+@@ -21,6 +21,7 @@ package cache
+ 
+ import (
+ 	"hash/maphash"
++	"sync/atomic"
+ 	"time"
+ 
+ 	"github.com/IrineSistiana/mosdns/v5/pkg/cache"
+@@ -78,6 +79,7 @@ type item struct {
+ 	resp           *dns.Msg
+ 	storedTime     time.Time
+ 	expirationTime time.Time
++	hitCount       atomic.Uint32
+ }
+ 
+ func copyNoOpt(m *dns.Msg) *dns.Msg {