1	/*
     2	Copyright 2011 The Perkeep Authors
     3	
     4	Licensed under the Apache License, Version 2.0 (the "License");
     5	you may not use this file except in compliance with the License.
     6	You may obtain a copy of the License at
     7	
     8	     http://www.apache.org/licenses/LICENSE-2.0
     9	
    10	Unless required by applicable law or agreed to in writing, software
    11	distributed under the License is distributed on an "AS IS" BASIS,
    12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13	See the License for the specific language governing permissions and
    14	limitations under the License.
    15	*/
    16	
    17	// Package cacher provides various blobref fetching caching mechanisms.
    18	package cacher // import "perkeep.org/pkg/cacher"
    19	
    20	import (
    21		"context"
    22		"io"
    23		"log"
    24		"os"
    25		"path/filepath"
    26	
    27		"perkeep.org/internal/osutil"
    28		"perkeep.org/pkg/blob"
    29		"perkeep.org/pkg/blobserver"
    30		"perkeep.org/pkg/blobserver/localdisk"
    31	
    32		"go4.org/syncutil/singleflight"
    33	)
    34	
    35	// NewCachingFetcher returns a CachingFetcher that fetches from
    36	// fetcher and writes to and serves from cache.
    37	func NewCachingFetcher(cache blobserver.Cache, fetcher blob.Fetcher) *CachingFetcher {
    38		return &CachingFetcher{c: cache, sf: fetcher}
    39	}
    40	
    41	// A CachingFetcher is a blob.Fetcher and a blob.SeekFetcher.
    42	type CachingFetcher struct {
    43		c  blobserver.Cache
    44		sf blob.Fetcher
    45		// cacheHitHook, if set, is called right after a cache hit. It is meant to add
    46		// potential side-effects from calling the Fetcher that would have happened
    47		// if we had had a cache miss. It is the responsibility of cacheHitHook to return
    48		// a ReadCloser equivalent to the state that rc was given in.
    49		cacheHitHook func(br blob.Ref, rc io.ReadCloser) (io.ReadCloser, error)
    50	
    51		g singleflight.Group
    52	}
    53	
    54	// SetCacheHitHook sets a function that will modify the return values from Fetch
    55	// in the case of a cache hit.
    56	// Its purpose is to add potential side-effects from calling the Fetcher that would
    57	// have happened if we had had a cache miss. It is the responsibility of fn to
    58	// return a ReadCloser equivalent to the state that rc was given in.
    59	func (cf *CachingFetcher) SetCacheHitHook(fn func(br blob.Ref, rc io.ReadCloser) (io.ReadCloser, error)) {
    60		cf.cacheHitHook = fn
    61	}
    62	
    63	func (cf *CachingFetcher) Fetch(ctx context.Context, br blob.Ref) (content io.ReadCloser, size uint32, err error) {
    64		content, size, err = cf.c.Fetch(ctx, br)
    65		if err == nil {
    66			if cf.cacheHitHook != nil {
    67				rc, err := cf.cacheHitHook(br, content)
    68				if err != nil {
    69					content.Close()
    70					return nil, 0, err
    71				}
    72				content = rc
    73			}
    74			return
    75		}
    76		if err = cf.faultIn(ctx, br); err != nil {
    77			return
    78		}
    79		return cf.c.Fetch(ctx, br)
    80	}
    81	
    82	func (cf *CachingFetcher) faultIn(ctx context.Context, br blob.Ref) error {
    83		_, err := cf.g.Do(br.String(), func() (interface{}, error) {
    84			sblob, _, err := cf.sf.Fetch(ctx, br)
    85			if err != nil {
    86				return nil, err
    87			}
    88			defer sblob.Close()
    89			_, err = blobserver.Receive(ctx, cf.c, br, sblob)
    90			return nil, err
    91		})
    92		return err
    93	}
    94	
    95	// A DiskCache is a blob.Fetcher that serves from a local temp
    96	// directory and is backed by a another blob.Fetcher (usually the
    97	// pkg/client HTTP client).
    98	type DiskCache struct {
    99		*CachingFetcher
   100	
   101		// Root is the temp directory being used to store files.
   102		// It is available mostly for debug printing.
   103		Root string
   104	
   105		cleanAll bool // cleaning policy. TODO: something better.
   106	}
   107	
   108	// NewDiskCache returns a new DiskCache from a Fetcher, which
   109	// is usually the pkg/client HTTP client (which typically has much
   110	// higher latency and lower bandwidth than local disk).
   111	func NewDiskCache(fetcher blob.Fetcher) (*DiskCache, error) {
   112		cacheDir := filepath.Join(osutil.CacheDir(), "blobs")
   113		if !osutil.DirExists(cacheDir) {
   114			if err := os.Mkdir(cacheDir, 0700); err != nil {
   115				log.Printf("Warning: failed to make %s: %v; using tempdir instead", cacheDir, err)
   116				cacheDir, err = os.MkdirTemp("", "camlicache")
   117				if err != nil {
   118					return nil, err
   119				}
   120			}
   121		}
   122		// TODO: max disk size, keep LRU of access, smarter cleaning, etc
   123		// TODO: use diskpacked instead? harder to clean, though.
   124		diskcache, err := localdisk.New(cacheDir)
   125		if err != nil {
   126			return nil, err
   127		}
   128		dc := &DiskCache{
   129			CachingFetcher: NewCachingFetcher(diskcache, fetcher),
   130			Root:           cacheDir,
   131		}
   132		return dc, nil
   133	}
   134	
   135	// Clean cleans some or all of the DiskCache.
   136	func (dc *DiskCache) Clean() {
   137		// TODO: something between nothing and deleting everything.
   138		if dc.cleanAll {
   139			os.RemoveAll(dc.Root)
   140		}
   141	}
   142	
   143	var (
   144		_ blob.Fetcher = (*CachingFetcher)(nil)
   145		_ blob.Fetcher = (*DiskCache)(nil)
   146	)