solver: release unreferenced cache keys after gc

control/control.go

@@ -76,14 +76,15 @@ type Opt struct {
 
 type Controller struct { // TODO: ControlService
 	// buildCount needs to be 64bit aligned
-	buildCount       int64
-	opt              Opt
-	solver           *llbsolver.Solver
-	history          *llbsolver.HistoryQueue
-	cache            solver.CacheManager
-	gatewayForwarder *controlgateway.GatewayForwarder
-	throttledGC      func()
-	gcmu             sync.Mutex
+	buildCount                   int64
+	opt                          Opt
+	solver                       *llbsolver.Solver
+	history                      *llbsolver.HistoryQueue
+	cache                        solver.CacheManager
+	gatewayForwarder             *controlgateway.GatewayForwarder
+	throttledGC                  func()
+	throttledReleaseUnreferenced func()
+	gcmu                         sync.Mutex
 	tracev1.UnimplementedTraceServiceServer
 }
 
@@ -124,6 +125,8 @@ func NewController(opt Opt) (*Controller, error) {
 		gatewayForwarder: gatewayForwarder,
 	}
 	c.throttledGC = throttle.After(time.Minute, c.gc)
+	// use longer interval for releaseUnreferencedCache deleting links quickly is less important
+	c.throttledReleaseUnreferenced = throttle.After(5*time.Minute, func() { c.releaseUnreferencedCache(context.TODO()) })
 
 	defer func() {
 		time.AfterFunc(time.Second, c.throttledGC)
@@ -194,6 +197,10 @@ func (c *Controller) DiskUsage(ctx context.Context, r *controlapi.DiskUsageReque
 	return resp, nil
 }
 
+func (c *Controller) releaseUnreferencedCache(ctx context.Context) error {
+	return c.cache.ReleaseUnreferenced(ctx)
+}
+
 func (c *Controller) Prune(req *controlapi.PruneRequest, stream controlapi.Control_PruneServer) error {
 	if atomic.LoadInt64(&c.buildCount) == 0 {
 		imageutil.CancelCacheLeases()
@@ -634,6 +641,7 @@ func (c *Controller) gc() {
 	<-done
 	if size > 0 {
 		bklog.G(ctx).Debugf("gc cleaned up %d bytes", size)
+		go c.throttledReleaseUnreferenced()
 	}
 }
 

solver/cachemanager.go

@@ -42,8 +42,13 @@ type cacheManager struct {
 }
 
 func (c *cacheManager) ReleaseUnreferenced(ctx context.Context) error {
+	visited := map[string]struct{}{}
 	return c.backend.Walk(func(id string) error {
 		return c.backend.WalkResults(id, func(cr CacheResult) error {
+			if _, ok := visited[cr.ID]; ok {
+				return nil
+			}
+			visited[cr.ID] = struct{}{}
 			if !c.results.Exists(ctx, cr.ID) {
 				c.backend.Release(cr.ID)
 			}

solver/combinedcache.go

@@ -33,6 +33,18 @@ func (cm *combinedCacheManager) ID() string {
 	return cm.id
 }
 
+func (cm *combinedCacheManager) ReleaseUnreferenced(ctx context.Context) error {
+	eg, ctx := errgroup.WithContext(ctx)
+	for _, c := range cm.cms {
+		func(c CacheManager) {
+			eg.Go(func() error {
+				return c.ReleaseUnreferenced(ctx)
+			})
+		}(c)
+	}
+	return eg.Wait()
+}
+
 func (cm *combinedCacheManager) Query(inp []CacheKeyWithSelector, inputIndex Index, dgst digest.Digest, outputIndex Index) ([]*CacheKey, error) {
 	eg, _ := errgroup.WithContext(context.TODO())
 	keys := make(map[string]*CacheKey, len(cm.cms))

solver/llbsolver/bridge.go

@@ -425,6 +425,13 @@ func (lcm *lazyCacheManager) Save(key *solver.CacheKey, s solver.Result, created
 	return lcm.main.Save(key, s, createdAt)
 }
 
+func (lcm *lazyCacheManager) ReleaseUnreferenced(ctx context.Context) error {
+	if err := lcm.wait(); err != nil {
+		return err
+	}
+	return lcm.main.ReleaseUnreferenced(ctx)
+}
+
 func (lcm *lazyCacheManager) wait() error {
 	<-lcm.waitCh
 	return lcm.err

solver/types.go

@@ -267,4 +267,6 @@ type CacheManager interface {
 
 	// Save saves a result based on a cache key
 	Save(key *CacheKey, s Result, createdAt time.Time) (*ExportableCacheKey, error)
+
+	ReleaseUnreferenced(context.Context) error
 }