diff --git a/cache/disk/disk.go b/cache/disk/disk.go index 2d7eb8d87..106d04dbc 100644 --- a/cache/disk/disk.go +++ b/cache/disk/disk.go @@ -31,6 +31,26 @@ var ( Name: "bazel_remote_disk_cache_misses", Help: "The total number of disk backend cache misses", }) + evictedUncommited = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bazel_remote_disk_cache_evicted_uncommited", + Help: "The total number of uncommited entries evicted from disk backend cache.", + }) + newAcceptedWhenExistingCommited = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bazel_remote_disk_cache_new_accepted_commited", + Help: "The total number of new disk backend put requests accepted when an already existing commited entry", + }) + newDiscardedWhenExistingCommited = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bazel_remote_disk_cache_new_discarded_commited", + Help: "The total number of new disk backend put requests discarded when an already existing commited entry", + }) + newAcceptedWhenExistingUncommited = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bazel_remote_disk_cache_new_accepted_uncommited", + Help: "The total number of new disk backend put requests accepted when an already existing uncommited entry", + }) + newDiscardedWhenExistingUncommited = promauto.NewCounter(prometheus.CounterOpts{ + Name: "bazel_remote_disk_cache_new_discarded_uncommited", + Help: "The total number of new disk backend put requests discarded when an already existing uncommited entry", + }) ) // lruItem is the type of the values stored in SizedLRU to keep track of items. @@ -51,6 +71,8 @@ type Cache struct { mu *sync.Mutex lru SizedLRU + readAfterWriteGuarantee bool + overwriteCommited bool } type nameAndInfo struct { @@ -63,7 +85,7 @@ const sha256HashStrSize = sha256.Size * 2 // Two hex characters per byte. // New returns a new instance of a filesystem-based cache rooted at `dir`, // with a maximum size of `maxSizeBytes` bytes and an optional backend `proxy`. // Cache is safe for concurrent use. -func New(dir string, maxSizeBytes int64, proxy cache.Proxy) *Cache { +func New(dir string, maxSizeBytes int64, proxy cache.Proxy, readAfterWriteGuarantee bool, overwriteCommited bool) *Cache { // Create the directory structure. hexLetters := []byte("0123456789abcdef") for _, c1 := range hexLetters { @@ -101,47 +123,24 @@ func New(dir string, maxSizeBytes int64, proxy cache.Proxy) *Cache { return } - // There is an ongoing upload for the evicted item. The temp - // file may or may not exist at this point. + // If there are concurrent uploads ongoing for a key that is + // evicted, then those uploads are allowed to continue and will + // re-add themself when finished. + // + // Trying to delete those files in uploading state would be + // problematic since the eviction might occur when they + // released the lock after they added as uncommited in LRU, + // but before they created the temporary file, resulting in + // that they might continue creating both their temporary file + // and the real file after onEvict already finished. // - // We should either be able to remove both the temp file and - // the regular cache file, or to remove just the regular cache - // file. The temp file is renamed/moved to the regular cache - // file without holding the lock, so we must try removing the - // temp file first. - // Note: if you hit this case, then your cache size might be // too small (blobs are moved to the most-recently used end // of the index when the upload begins, and these items are // still uploading when they reach the least-recently used // end of the index). - tf := f + ".tmp" - var fErr, tfErr error - removedCount := 0 - - tfErr = os.Remove(tf) - if tfErr == nil { - removedCount++ - } - - fErr = os.Remove(f) - if fErr == nil { - removedCount++ - } - - // We expect to have removed at least one file at this point. - if removedCount == 0 { - if !os.IsNotExist(tfErr) { - log.Printf("ERROR: failed to remove evicted item: %s / %v", - tf, tfErr) - } - - if !os.IsNotExist(fErr) { - log.Printf("ERROR: failed to remove evicted item: %s / %v", - f, fErr) - } - } + evictedUncommited.Inc() } c := &Cache{ @@ -149,8 +148,12 @@ func New(dir string, maxSizeBytes int64, proxy cache.Proxy) *Cache { proxy: proxy, mu: &sync.Mutex{}, lru: NewSizedLRU(maxSizeBytes, onEvict), + readAfterWriteGuarantee: readAfterWriteGuarantee, + overwriteCommited: overwriteCommited, } + log.Printf("Read after write guarantee: %t\n", c.readAfterWriteGuarantee) + log.Printf("Overwrite commited: %t\n", c.overwriteCommited) err := c.migrateDirectories() if err != nil { log.Fatalf("Attempting to migrate the old directory structure to the new structure failed "+ @@ -261,24 +264,68 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. c.mu.Lock() - // If there's an ongoing upload (i.e. cache key is present in uncommitted state), - // we drop the upload and discard the incoming stream. We do accept uploads - // of existing keys, as it should happen relatively rarely (e.g. race - // condition on the bazel side) but it's useful to overwrite poisoned items. - if existingItem, found := c.lru.Get(key); found { - if !existingItem.(*lruItem).committed { - c.mu.Unlock() - io.Copy(ioutil.Discard, r) - return nil - } - } - // Try to add the item to the LRU. newItem := &lruItem{ size: expectedSize, committed: false, } - ok := c.lru.Add(key, newItem) + ok := true + if existingItem, found := c.lru.Get(key); found { + if existingItem.(*lruItem).committed { + if c.overwriteCommited { + // Original bazel-remote had this behavour of accepting uploads + // of existing commited keys, and motivated it by it would + // happen relatively rarely (e.g. race condition on bazel side) + // and that it is useful to overwrite poisoned items. + newAcceptedWhenExistingCommited.Inc() + ok = c.lru.HasValidSize(newItem) + } else { + // Ignore new uploads of already existing keys. + // No need to consider readAfterWriteGuarantee + // because the original entry is already + // readable. + // + // Slightly more efficient, at least for remote execution + // use cases where bazel client uploads the same input + // files many times in paralllell. + newDiscardedWhenExistingCommited.Inc() + c.mu.Unlock() + io.Copy(ioutil.Discard, r) + return nil + } + } else { + if c.readAfterWriteGuarantee { + // Accept concurrent put requests. Handle all of + // them in parallell instead of having them + // wait for the first one to finish, in + // order to avoid that a fast uploader have + // to wait for a slow one. And also to avoid + // need for timeput handling if the first + // never finish. + newAcceptedWhenExistingUncommited.Inc() + ok = c.lru.HasValidSize(newItem) + } else { + // If ongoing upload (i.e. cache key is present + // in uncommitted state), we drop the upload, + // discard the incoming stream, and assume the + // other upload will soon finish successfully. + // + // Discarding is more efficient and often good + // enought when not needing guarantee about + // entries being readable directly after put + // request finish. However such guarantee is + // required when used as remote execution CAS + // and for bazel's builds-without-the-bytes feature. + newDiscardedWhenExistingUncommited.Inc() + c.mu.Unlock() + io.Copy(ioutil.Discard, r) + return nil + } + } + } else { + ok = c.lru.Add(key, newItem) + } + c.mu.Unlock() if !ok { return &cache.Error{ @@ -293,11 +340,50 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. shouldCommit := false filePath := cacheFilePath(kind, c.dir, hash) defer func() { + c.mu.Lock() - if shouldCommit { - newItem.committed = true + if c.lru.HasInstance(key, newItem) { + if shouldCommit { + // This is the normal and fast path. + newItem.committed = true + // It could be that another concurrent put had just + // overwritten our file before we mark it's key as + // commited here, but if so that other will soon + // re-add the file with the most recently written + // size. + } else { + c.lru.Remove(key) + } } else { - c.lru.Remove(key) + if shouldCommit { + // Three ways to end up here: + // a. Overwriting an already existing commited entry. + // b. Concurrent put requests with same key. + // c. Key evicted during put request of same key. + // + // Calling os.Stat while holding the lock should be acceptable since this + // is not the normal path, so performance is less critical. + fileStat, err := os.Stat(filePath) + if err == nil { + // There is no lock around file renaming, so we do not now if + // the current file on disk has been written by us or some + // other concurrent executing put request with the same key. + // Therefore get the correct size now while we are holding the + // lock to make sure we tell lru the correct size regardless of + // if it was written by us or not. + newItem.committed = true + newItem.size = fileStat.Size() + ok := c.lru.Add(key, newItem) + if !ok { + os.Remove(filePath) + } + } else { + // Our file has been evicted. Nothing needs to be done since + // we already verified this instance is no longer in LRU. + } + } else { + // We are not in LRU, and since upload failed, that how it should be. + } } c.mu.Unlock() @@ -310,12 +396,14 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. } }() - // Download to a temporary file - tmpFilePath := filePath + ".tmp" - f, err := os.Create(tmpFilePath) + // Download to a temporary file. Important with unique file name + // since concurrent uploads can go on in parallell. + f, err := ioutil.TempFile(cacheDirPath(kind, c.dir, hash), ".tmp") if err != nil { return err } + tmpFilePath := f.Name() + defer func() { if !shouldCommit { // Only delete the temp file if moving it didn't succeed. @@ -366,6 +454,7 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. return err } + // Return two bools, `available` is true if the item is in the local // cache and ready to use. // @@ -388,7 +477,7 @@ func (c *Cache) availableOrTryProxy(key string) (available bool, tryProxy bool) // Reserve a place in the LRU. // The caller must replace or remove this! tryProxy = c.lru.Add(key, &lruItem{ - size: 0, + size: 0, committed: false, }) } @@ -593,6 +682,10 @@ func cacheFilePath(kind cache.EntryKind, cacheDir string, hash string) string { return filepath.Join(cacheDir, cacheKey(kind, hash)) } +func cacheDirPath(kind cache.EntryKind, cacheDir string, hash string) string { + return filepath.Join(cacheDir, kind.String(), hash[:2]) +} + // GetValidatedActionResult returns a valid ActionResult and its serialized // value from the CAS if it and all its dependencies are also available. If // not, nil values are returned. If something unexpected went wrong, return diff --git a/cache/disk/lru.go b/cache/disk/lru.go index ab3f0120b..239d9a7e4 100644 --- a/cache/disk/lru.go +++ b/cache/disk/lru.go @@ -26,9 +26,11 @@ type SizedLRU interface { Add(key Key, value sizedItem) (ok bool) Get(key Key) (value sizedItem, ok bool) Remove(key Key) + HasInstance(key Key, value sizedItem) (has bool) Len() int CurrentSize() int64 MaxSize() int64 + HasValidSize(value sizedItem) (ok bool) } type sizedLRU struct { @@ -61,7 +63,7 @@ func NewSizedLRU(maxSize int64, onEvict EvictCallback) SizedLRU { // Add adds a (key, value) to the cache, evicting items as necessary. Add returns false ( // and does not add the item) if the item size is larger than the maximum size of the cache. func (c *sizedLRU) Add(key Key, value sizedItem) (ok bool) { - if value.Size() > c.maxSize { + if !c.HasValidSize(value) { return false } @@ -107,6 +109,15 @@ func (c *sizedLRU) Remove(key Key) { } } +// Returns true if a specific value instance is available via key +func (c *sizedLRU) HasInstance(key Key, value sizedItem) (has bool) { + if ee, ok := c.cache[key]; ok { + return value == ee.Value.(*entry).value + } else { + return false + } +} + // Len returns the number of items in the cache func (c *sizedLRU) Len() int { return len(c.cache) @@ -120,6 +131,11 @@ func (c *sizedLRU) MaxSize() int64 { return c.maxSize } + +func (c *sizedLRU) HasValidSize(value sizedItem) (ok bool) { + return value.Size() <= c.maxSize +} + func (c *sizedLRU) removeElement(e *list.Element) { c.ll.Remove(e) kv := e.Value.(*entry) diff --git a/config/config.go b/config/config.go index 452f16eba..9be920b42 100644 --- a/config/config.go +++ b/config/config.go @@ -52,6 +52,8 @@ type Config struct { IdleTimeout time.Duration `yaml:"idle_timeout"` DisableHTTPACValidation bool `yaml:"disable_http_ac_validation"` DisableGRPCACDepsCheck bool `yaml:"disable_grpc_ac_deps_check"` + ReadAfterWriteGuarantee bool `yaml:"read_after_write_guarantee"` + OverwriteCommited bool `yaml:"overwrite_commited"` } // New returns a validated Config with the specified values, and an error @@ -78,6 +80,8 @@ func New(dir string, maxSize int, host string, port int, grpcPort int, IdleTimeout: idleTimeout, DisableHTTPACValidation: disableHTTPACValidation, DisableGRPCACDepsCheck: disableGRPCACDepsCheck, + ReadAfterWriteGuarantee: false, + OverwriteCommited: true, } err := validateConfig(&c) diff --git a/main.go b/main.go index 0086c2ed2..4790a80ff 100644 --- a/main.go +++ b/main.go @@ -281,7 +281,8 @@ func main() { proxyCache = s3proxy.New(c.S3CloudStorage, accessLogger, errorLogger) } - diskCache := disk.New(c.Dir, int64(c.MaxSize)*1024*1024*1024, proxyCache) + diskCache := disk.New(c.Dir, int64(c.MaxSize)*1024*1024*1024, proxyCache, + c.ReadAfterWriteGuarantee, c.OverwriteCommited) mux := http.NewServeMux() httpServer := &http.Server{