diff --git a/README.md b/README.md index 059dcb9d1..922a98b7f 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ Returns the cache status/info. $ curl http://localhost:8080/status { "CurrSize": 414081715503, + "ReservedSize": 876400, "MaxSize": 8589934592000, "NumFiles": 621413, "ServerTime": 1588329927, diff --git a/cache/disk/BUILD.bazel b/cache/disk/BUILD.bazel index f4006afef..96a899dd5 100644 --- a/cache/disk/BUILD.bazel +++ b/cache/disk/BUILD.bazel @@ -10,6 +10,7 @@ go_library( visibility = ["//visibility:public"], deps = [ "//cache:go_default_library", + "//utils/tempfile:go_default_library", "@com_github_bazelbuild_remote_apis//build/bazel/remote/execution/v2:go_default_library", "@com_github_djherbis_atime//:go_default_library", "@com_github_golang_protobuf//proto:go_default_library", diff --git a/cache/disk/disk.go b/cache/disk/disk.go index 40f9a6e98..180511281 100644 --- a/cache/disk/disk.go +++ b/cache/disk/disk.go @@ -14,8 +14,11 @@ import ( "regexp" "sort" "sync" + "syscall" "github.com/buchgr/bazel-remote/cache" + "github.com/buchgr/bazel-remote/utils/tempfile" + "github.com/djherbis/atime" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" @@ -35,11 +38,12 @@ var ( }) ) +var tfc = tempfile.NewCreator() + // lruItem is the type of the values stored in SizedLRU to keep track of items. // It implements the sizedItem interface. type lruItem struct { - size int64 - committed bool + size int64 } func (i *lruItem) Size() int64 { @@ -91,59 +95,10 @@ func New(dir string, maxSizeBytes int64, proxy cache.Proxy) *Cache { // This function is only called while the lock is held // by the current goroutine. onEvict := func(key Key, value sizedItem) { - f := filepath.Join(dir, key.(string)) - - if value.(*lruItem).committed { - // Common case. Just remove the cache file and we're done. - err := os.Remove(f) - if err != nil { - log.Printf("ERROR: failed to remove evicted cache file: %s", f) - } - - return - } - - // There is an ongoing upload for the evicted item. The temp - // file may or may not exist at this point. - // - // 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) - } + err := os.Remove(f) + if err != nil { + log.Printf("ERROR: failed to remove evicted cache file: %s", f) } } @@ -259,10 +214,7 @@ func (c *Cache) loadExistingFiles() error { log.Println("Building LRU index.") for _, f := range files { relPath := f.name[len(c.dir)+1:] - ok := c.lru.Add(relPath, &lruItem{ - size: f.info.Size(), - committed: true, - }) + ok := c.lru.Add(relPath, &lruItem{size: f.info.Size()}) if !ok { err = os.Remove(filepath.Join(c.dir, relPath)) if err != nil { @@ -275,10 +227,13 @@ func (c *Cache) loadExistingFiles() error { return nil } -// Put stores a stream of `expectedSize` bytes from `r` into the cache. +// Put stores a stream of `size` bytes from `r` into the cache. // If `hash` is not the empty string, and the contents don't match it, // a non-nil error is returned. -func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io.Reader) error { +func (c *Cache) Put(kind cache.EntryKind, hash string, size int64, r io.Reader) (rErr error) { + if size < 0 { + return fmt.Errorf("Invalid (negative) size: %d", size) + } // The hash format is checked properly in the http/grpc code. // Just perform a simple/fast check here, to catch bad tests. @@ -287,90 +242,111 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. len(hash), sha256.Size) } - if kind == cache.CAS && expectedSize == 0 && hash == emptySha256 { + if kind == cache.CAS && size == 0 && hash == emptySha256 { io.Copy(ioutil.Discard, r) return nil } key := cacheKey(kind, hash) - c.mu.Lock() + var tf *os.File // Tempfile. - // 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 + // Cleanup intermediate state if something went wrong and we + // did not successfully commit. + unreserve := false + removeTempfile := false + defer func() { + // No lock required to remove stray tempfiles. + if removeTempfile { + os.Remove(tf.Name()) } - } - // Try to add the item to the LRU. - newItem := &lruItem{ - size: expectedSize, - committed: false, - } - ok := c.lru.Add(key, newItem) - c.mu.Unlock() - if !ok { - return &cache.Error{ - Code: http.StatusInsufficientStorage, - Text: fmt.Sprintf("The item is larger (%d) than the cache's maximum size (%d).", - expectedSize, c.lru.MaxSize()), + if unreserve { + c.mu.Lock() + err := c.lru.Unreserve(size) + if err != nil { + // Set named return value. + rErr = err + log.Printf(rErr.Error()) + } + c.mu.Unlock() } - } + }() - // By the time this function exits, we should either mark the LRU item as committed - // (if the upload went well), or delete it. Capturing the flag variable is not very nice, - // but this stuff is really easy to get wrong without defer(). - shouldCommit := false - filePath := cacheFilePath(kind, c.dir, hash) - defer func() { + if size > 0 { c.mu.Lock() - if shouldCommit { - newItem.committed = true - } else { - c.lru.Remove(key) + ok, err := c.lru.Reserve(size) + if err != nil { + c.mu.Unlock() + return &cache.Error{ + Code: http.StatusInternalServerError, + Text: err.Error(), + } } - c.mu.Unlock() - - if shouldCommit && c.proxy != nil { - // TODO: buffer in memory, avoid a filesystem round-trip? - fr, err := os.Open(filePath) - if err == nil { - c.proxy.Put(kind, hash, expectedSize, fr) + if !ok { + c.mu.Unlock() + return &cache.Error{ + Code: http.StatusInsufficientStorage, + Text: fmt.Sprintf("The item (%d) + reserved space is larger than the cache's maximum size (%d).", + size, c.lru.MaxSize()), } } - }() + c.mu.Unlock() + unreserve = true + } - // Download to a temporary file - tmpFilePath := filePath + ".tmp" - f, err := os.Create(tmpFilePath) + // Final destination, if all goes well. + filePath := cacheFilePath(kind, c.dir, hash) + + // We will download to this temporary file. + tf, err := tfc.Create(filePath) if err != nil { return err } + removeTempfile = true + + err = writeAndCloseFile(r, kind, hash, size, tf) + if err != nil { + return err + } + + if c.proxy != nil { + rc, err := os.Open(tf.Name()) + if err != nil { + log.Println("Failed to proxy Put:", err) + } else { + // Doesn't block, should be fast. + c.proxy.Put(kind, hash, size, rc) + } + } + + unreserve, removeTempfile, err = c.commit(key, tf.Name(), filePath, size, size) + + // Might be nil. + return err +} + +func writeAndCloseFile(r io.Reader, kind cache.EntryKind, hash string, size int64, f *os.File) error { + closeFile := true defer func() { - if !shouldCommit { - // Only delete the temp file if moving it didn't succeed. - os.Remove(tmpFilePath) + if closeFile { + f.Close() } - // Just in case we didn't already close it. No need to check errors. - f.Close() }() + var err error + var bytesCopied int64 if kind == cache.CAS { hasher := sha256.New() - if bytesCopied, err = io.Copy(io.MultiWriter(f, hasher), r); err != nil { + bytesCopied, err = io.Copy(io.MultiWriter(f, hasher), r) + if err != nil { return err } actualHash := hex.EncodeToString(hasher.Sum(nil)) if actualHash != hash { return fmt.Errorf( - "hashsums don't match. Expected %s, found %s", key, actualHash) + "checksums don't match. Expected %s, found %s", hash, actualHash) } } else { if bytesCopied, err = io.Copy(f, r); err != nil { @@ -378,6 +354,11 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. } } + if isSizeMismatch(bytesCopied, size) { + return fmt.Errorf( + "sizes don't match. Expected %d, found %d", size, bytesCopied) + } + if err = f.Sync(); err != nil { return err } @@ -385,55 +366,104 @@ func (c *Cache) Put(kind cache.EntryKind, hash string, expectedSize int64, r io. if err = f.Close(); err != nil { return err } + closeFile = false - if bytesCopied != expectedSize { - return fmt.Errorf( - "sizes don't match. Expected %d, found %d", expectedSize, bytesCopied) + return nil +} + +// This must be called when the lock is not held. +func (c *Cache) commit(key string, tempfile string, finalPath string, reservedSize int64, foundSize int64) (unreserve bool, removeTempfile bool, err error) { + unreserve = reservedSize > 0 + removeTempfile = true + + c.mu.Lock() + defer c.mu.Unlock() + + if unreserve { + err = c.lru.Unreserve(reservedSize) + if err != nil { + log.Println(err.Error()) + return true, removeTempfile, err + } } + unreserve = false - // Rename to the final path - err = os.Rename(tmpFilePath, filePath) - if err == nil { - // Only commit if renaming succeeded. - // This flag is used by the defer() block above. - shouldCommit = true + if !c.lru.Add(key, &lruItem{size: foundSize}) { + err = fmt.Errorf("INTERNAL ERROR: failed to add: %s, size %d", key, foundSize) + log.Println(err.Error()) + return unreserve, removeTempfile, err } - return err -} + err = syscall.Rename(tempfile, finalPath) + if err != nil { + log.Printf("INTERNAL ERROR: failed to rename \"%s\" to \"%s\": %v", + tempfile, finalPath, err) + log.Println("Removing", key) + c.lru.Remove(key) + return unreserve, removeTempfile, err + } + removeTempfile = false -// Return two bools, `available` is true if the item is in the local -// cache and ready to use. -// -// `tryProxy` is true if the item is not in the local cache but can -// be requested from the proxy, in which case, a placeholder entry -// has been added to the index and the caller must either replace -// the entry with the actual size, or remove it from the LRU. -func (c *Cache) availableOrTryProxy(key string) (available bool, tryProxy bool) { - inProgress := false - tryProxy = false + // Commit successful if we made it this far! \o/ + return unreserve, removeTempfile, nil +} +// Return a non-nil *os.File and non-negative size if the item is available +// locally, and a boolean that indicates if the item is not available locally +// but that we can try the proxy backend. +func (c *Cache) availableOrTryProxy(key string, size int64, blobPath string) (rc io.ReadCloser, foundSize int64, tryProxy bool, err error) { + locked := true c.mu.Lock() - existingItem, found := c.lru.Get(key) - if found { - if !existingItem.(*lruItem).committed { - inProgress = true + val, available := c.lru.Get(key) + if available { + c.mu.Unlock() // We expect a cache hit below. + locked = false + + item := val.(*lruItem) + if !isSizeMismatch(size, item.size) { + var f *os.File + f, err = os.Open(blobPath) + if err != nil { + // Race condition, was the item purged after we released the lock? + log.Printf("Warning: expected %s to exist on disk, undersized cache?", blobPath) + } else { + var fileInfo os.FileInfo + fileInfo, err = f.Stat() + foundSize := fileInfo.Size() + if isSizeMismatch(size, foundSize) { + // Race condition, was the item replaced after we released the lock? + log.Printf("Warning: expected %s to on disk to have size %d, found %d", + blobPath, size, foundSize) + } else { + return f, foundSize, false, nil + } + } } - } else if c.proxy != nil { - // Reserve a place in the LRU. - // The caller must replace or remove this! - tryProxy = c.lru.Add(key, &lruItem{ - size: 0, - committed: false, - }) } + err = nil - c.mu.Unlock() + if c.proxy != nil { + if size > 0 { + // If we know the size, attempt to reserve that much space. + if !locked { + c.mu.Lock() + } + tryProxy, err = c.lru.Reserve(size) + c.mu.Unlock() + locked = false + } else { + // If the size is unknown, take a risk and hope it's not + // too large. + tryProxy = true + } + } - available = found && !inProgress + if locked { + c.mu.Unlock() + } - return available, tryProxy + return nil, -1, tryProxy, err } // Get returns an io.ReadCloser with the content of the cache item stored @@ -441,7 +471,7 @@ func (c *Cache) availableOrTryProxy(key string) (available bool, tryProxy bool) // item is not found, the io.ReadCloser will be nil. If some error occurred // when processing the request, then it is returned. Callers should provide // the `size` of the item to be retrieved, or -1 if unknown. -func (c *Cache) Get(kind cache.EntryKind, hash string, size int64) (io.ReadCloser, int64, error) { +func (c *Cache) Get(kind cache.EntryKind, hash string, size int64) (rc io.ReadCloser, s int64, rErr error) { // The hash format is checked properly in the http/grpc code. // Just perform a simple/fast check here, to catch bad tests. @@ -458,29 +488,42 @@ func (c *Cache) Get(kind cache.EntryKind, hash string, size int64) (io.ReadClose var err error key := cacheKey(kind, hash) - available, tryProxy := c.availableOrTryProxy(key) + var tf *os.File // Tempfile we will write to. - if available { - blobPath := cacheFilePath(kind, c.dir, hash) - var fileInfo os.FileInfo - fileInfo, err = os.Stat(blobPath) - if err == nil { - foundSize := fileInfo.Size() - if isSizeMismatch(size, foundSize) { - cacheMisses.Inc() - return nil, -1, nil - } - var f *os.File - f, err = os.Open(blobPath) - if err == nil { - cacheHits.Inc() - return f, foundSize, nil + // Cleanup intermediate state if something went wrong and we + // did not successfully commit. + unreserve := false + removeTempfile := false + defer func() { + // No lock required to remove stray tempfiles. + if removeTempfile { + os.Remove(tf.Name()) + } + + if unreserve { + c.mu.Lock() + err := c.lru.Unreserve(size) + if err != nil { + // Set named return value. + rErr = err + log.Printf(rErr.Error()) } + c.mu.Unlock() } + }() - cacheMisses.Inc() + blobPath := cacheFilePath(kind, c.dir, hash) + f, foundSize, tryProxy, err := c.availableOrTryProxy(key, size, blobPath) + if err != nil { return nil, -1, err } + if tryProxy && size > 0 { + unreserve = true + } + if f != nil { + cacheHits.Inc() + return f, foundSize, nil + } cacheMisses.Inc() @@ -488,41 +531,6 @@ func (c *Cache) Get(kind cache.EntryKind, hash string, size int64) (io.ReadClose return nil, -1, nil } - filePath := cacheFilePath(kind, c.dir, hash) - tmpFilePath := filePath + ".tmp" - shouldCommit := false - tmpFileCreated := false - foundSize := int64(-1) - var f *os.File - - // We're allowed to try downloading this blob from the proxy. - // Before returning, we have to either commit the item and set - // its size, or remove the item from the LRU. - defer func() { - c.mu.Lock() - - if shouldCommit { - // Overwrite the placeholder inserted by availableOrTryProxy. - // Call Add instead of updating the entry directly, so we - // update the currentSize value. - c.lru.Add(key, &lruItem{ - size: foundSize, - committed: true, - }) - } else { - // Remove the placeholder. - c.lru.Remove(key) - } - - c.mu.Unlock() - - if !shouldCommit && tmpFileCreated { - os.Remove(tmpFilePath) // No need to check the error. - } - - f.Close() // No need to check the error. - }() - r, foundSize, err := c.proxy.Get(kind, hash) if r != nil { defer r.Close() @@ -534,44 +542,30 @@ func (c *Cache) Get(kind cache.EntryKind, hash string, size int64) (io.ReadClose return nil, -1, nil } - f, err = os.Create(tmpFilePath) + tf, err = tfc.Create(blobPath) if err != nil { return nil, -1, err } - tmpFileCreated = true + removeTempfile = true - written, err := io.Copy(f, r) + err = writeAndCloseFile(r, kind, hash, foundSize, tf) if err != nil { return nil, -1, err } - if written != foundSize { - return nil, -1, err - } - - if err = f.Sync(); err != nil { - return nil, -1, err - } - - if err = f.Close(); err != nil { + rc, err = os.Open(tf.Name()) + if err != nil { return nil, -1, err } - // Rename to the final path - err = os.Rename(tmpFilePath, filePath) - if err == nil { - // Only commit if renaming succeeded. - // This flag is used by the defer() block above. - shouldCommit = true - - var f2 *os.File - f2, err = os.Open(filePath) - if err == nil { - return f2, foundSize, nil - } + unreserve, removeTempfile, err = c.commit(key, tf.Name(), blobPath, size, foundSize) + if err != nil { + rc.Close() + rc = nil + foundSize = -1 } - return nil, -1, err + return rc, foundSize, err } // Contains returns true if the `hash` key exists in the cache, and @@ -593,33 +587,26 @@ func (c *Cache) Contains(kind cache.EntryKind, hash string, size int64) (bool, i return true, 0 } - var found bool foundSize := int64(-1) key := cacheKey(kind, hash) c.mu.Lock() - val, isInLru := c.lru.Get(key) - // Uncommitted (i.e. uploading items) should be reported as not ok - if isInLru { + val, exists := c.lru.Get(key) + if exists { item := val.(*lruItem) - found = item.committed foundSize = item.size } c.mu.Unlock() - if found { - if isSizeMismatch(size, foundSize) { - return false, -1 - } + if exists && !isSizeMismatch(size, foundSize) { return true, foundSize } if c.proxy != nil { - found, foundSize = c.proxy.Contains(kind, hash) - if isSizeMismatch(size, foundSize) { - return false, -1 + exists, foundSize = c.proxy.Contains(kind, hash) + if exists && !isSizeMismatch(size, foundSize) { + return true, foundSize } - return found, foundSize } return false, -1 @@ -633,11 +620,11 @@ func (c *Cache) MaxSize() int64 { // Stats returns the current size of the cache in bytes, and the number of // items stored in the cache. -func (c *Cache) Stats() (currentSize int64, numItems int) { +func (c *Cache) Stats() (totalSize int64, reservedSize int64, numItems int) { c.mu.Lock() defer c.mu.Unlock() - return c.lru.CurrentSize(), c.lru.Len() + return c.lru.TotalSize(), c.lru.ReservedSize(), c.lru.Len() } func isSizeMismatch(requestedSize int64, foundSize int64) bool { diff --git a/cache/disk/disk_test.go b/cache/disk/disk_test.go index bcb3a642d..d4b341529 100644 --- a/cache/disk/disk_test.go +++ b/cache/disk/disk_test.go @@ -37,15 +37,8 @@ func checkItems(cache *Cache, expSize int64, expNum int) error { if cache.lru.Len() != expNum { return fmt.Errorf("expected %d files in the cache, found %d", expNum, cache.lru.Len()) } - if cache.lru.CurrentSize() != expSize { - return fmt.Errorf("expected %d bytes in the cache, found %d", expSize, cache.lru.CurrentSize()) - } - - // Dig into the internals of the cache to make sure that all items are committed. - for _, it := range cache.lru.(*sizedLRU).cache { - if it.Value.(*entry).value.(*lruItem).committed != true { - return fmt.Errorf("expected committed = true") - } + if cache.lru.TotalSize() != expSize { + return fmt.Errorf("expected %d bytes in the cache, found %d", expSize, cache.lru.TotalSize()) } numFiles := 0 @@ -482,7 +475,7 @@ func TestMigrateFromOldDirectoryStructure(t *testing.T) { t.Fatal(err) } testCache := New(cacheDir, 2560, nil) - _, numItems := testCache.Stats() + _, _, numItems := testCache.Stats() if numItems != 3 { t.Fatalf("Expected test cache size 3 but was %d", numItems) } @@ -526,7 +519,7 @@ func TestLoadExistingEntries(t *testing.T) { } testCache := New(cacheDir, blobSize*numBlobs, nil) - _, numItems := testCache.Stats() + _, _, numItems := testCache.Stats() if int64(numItems) != numBlobs { t.Fatalf("Expected test cache size %d but was %d", numBlobs, numItems) @@ -582,7 +575,7 @@ func TestDistinctKeyspaces(t *testing.T) { t.Fatal(err) } - _, numItems := testCache.Stats() + _, _, numItems := testCache.Stats() if numItems != 3 { t.Fatalf("Expected test cache size 3 but was %d", numItems) diff --git a/cache/disk/lru.go b/cache/disk/lru.go index ab3f0120b..77a1079ea 100644 --- a/cache/disk/lru.go +++ b/cache/disk/lru.go @@ -2,6 +2,8 @@ package disk import ( "container/list" + "errors" + "fmt" ) type sizedItem interface { @@ -27,19 +29,32 @@ type SizedLRU interface { Get(key Key) (value sizedItem, ok bool) Remove(key Key) Len() int - CurrentSize() int64 + TotalSize() int64 + ReservedSize() int64 MaxSize() int64 + + // Reserve some amount of cache space. + Reserve(size int64) (ok bool, err error) + + // Release some amount of reserved cache space. + Unreserve(size int64) error } type sizedLRU struct { // Eviction double-linked list. Most recently accessed elements are at the front. ll *list.List // Map to access the items in O(1) time - cache map[interface{}]*list.Element + cache map[interface{}]*list.Element + + // Includes reserved size. currentSize int64 + + reservedSize int64 + // SizedLRU will evict items as needed to maintain the total size of the cache // below maxSize. maxSize int64 + onEvict EvictCallback } @@ -112,14 +127,88 @@ func (c *sizedLRU) Len() int { return len(c.cache) } -func (c *sizedLRU) CurrentSize() int64 { +func (c *sizedLRU) TotalSize() int64 { return c.currentSize } +func (c *sizedLRU) ReservedSize() int64 { + return c.reservedSize +} + func (c *sizedLRU) MaxSize() int64 { return c.maxSize } +// This assumes that a is positive, b is non-negative, and c is positive. +func sumLargerThan(a, b, c int64) bool { + sum := a + b + if sum > c { + return true + } + + if sum <= 0 { + // This indicates int64 overflow occurred. + return true + } + + return false +} + +var errReservation = errors.New("internal reservation error") + +func (c *sizedLRU) Reserve(size int64) (bool, error) { + if size == 0 { + return true, nil + } + + if size < 0 || size > c.maxSize { + return false, nil + } + + if sumLargerThan(size, c.reservedSize, c.maxSize) { + // If size + c.reservedSize is larger than c.maxSize + // then we cannot evict enough items to make enough + // space. + return false, nil + } + + // Evict elements until we are able to reserve enough space. + for sumLargerThan(size, c.currentSize, c.maxSize) { + ele := c.ll.Back() + if ele != nil { + c.removeElement(ele) + } else { + return false, errReservation // This should have been caught at the start. + } + } + + c.currentSize += size + c.reservedSize += size + return true, nil +} + +func (c *sizedLRU) Unreserve(size int64) error { + if size == 0 { + return nil + } + + if size < 0 { + return fmt.Errorf("INTERNAL ERROR: should not try to unreserve negative value: %d", size) + } + + newC := c.currentSize - size + newR := c.reservedSize - size + + if newC < 0 || newR < 0 { + return fmt.Errorf("INTERNAL ERROR: failed to unreserve: %d", size) + } + + c.currentSize = newC + c.reservedSize = newR + + return nil +} + func (c *sizedLRU) removeElement(e *list.Element) { c.ll.Remove(e) kv := e.Value.(*entry) diff --git a/cache/disk/lru_test.go b/cache/disk/lru_test.go index c1ea363d5..2693916af 100644 --- a/cache/disk/lru_test.go +++ b/cache/disk/lru_test.go @@ -2,6 +2,7 @@ package disk import ( "fmt" + "math" "reflect" "testing" ) @@ -16,9 +17,9 @@ func (it *testSizedItem) Size() int64 { } func checkSizeAndNumItems(t *testing.T, lru SizedLRU, expSize int64, expNum int) { - currentSize := lru.CurrentSize() + currentSize := lru.TotalSize() if currentSize != expSize { - t.Fatalf("CurrentSize: expected %d, got %d", expSize, currentSize) + t.Fatalf("TotalSize: expected %d, got %d", expSize, currentSize) } numItems := lru.Len() @@ -120,3 +121,136 @@ func TestRejectBigItem(t *testing.T) { checkSizeAndNumItems(t, lru, 0, 0) } + +func TestReserveSharesSpace(t *testing.T) { + largeItem := testSizedItem{math.MaxInt64, "pretend large item"} + + lru := NewSizedLRU(math.MaxInt64, nil) + lru.Add("foo", &largeItem) + ok, err := lru.Reserve(0) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatal("Should always be able to reserve 0") + } +} + +func TestReserveAtCapacity(t *testing.T) { + var ok bool + var err error + + lru := NewSizedLRU(math.MaxInt64, nil) + + ok, err = lru.Reserve(math.MaxInt64) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatal("Should be able to reserve all the space") + } + if lru.TotalSize() != math.MaxInt64 { + t.Fatalf("Expected total size %d, actual size %d", math.MaxInt64, + lru.TotalSize()) + } + + ok, err = lru.Reserve(0) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatal("Should always be able to reserve 0") + } + if lru.TotalSize() != math.MaxInt64 { + t.Fatalf("Expected total size %d, actual size %d", math.MaxInt64, + lru.TotalSize()) + } + + ok, err = lru.Reserve(1) + if err != nil { + t.Fatal(err) + } + if ok { + t.Fatal("Should not be able to reserve any space") + } + if lru.TotalSize() != math.MaxInt64 { + t.Fatalf("Expected total size %d, actual size %d", math.MaxInt64, + lru.TotalSize()) + } +} + +func TestReserveOverflow(t *testing.T) { + var lru SizedLRU + var ok bool + var err error + + lru = NewSizedLRU(1, nil) + + ok, err = lru.Reserve(1) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatalf("Expected to be able to reserve 1") + } + + ok, err = lru.Reserve(math.MaxInt64) + if err != nil { + t.Fatal(err) + } + if ok { + t.Fatal("Expected overflow") + } + + lru = NewSizedLRU(10, nil) + ok, err = lru.Reserve(math.MaxInt64) + if err != nil { + t.Fatal(err) + } + if ok { + t.Fatal("Expected overflow") + } +} + +func TestUnreserve(t *testing.T) { + var ok bool + var err error + + cap := int64(10) + lru := NewSizedLRU(cap, nil) + + for i := int64(1); i <= cap; i++ { + ok, err = lru.Reserve(1) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatal("Expected to be able to reserve 1") + } + if lru.TotalSize() != i { + t.Fatalf("Expected total size %d, actual size %d", i, + lru.TotalSize()) + } + } + + if lru.TotalSize() != cap { + t.Fatalf("Expected total size %d, actual size %d", cap, + lru.TotalSize()) + } + + for i := cap; i > 0; i-- { + err = lru.Unreserve(1) + if err != nil { + t.Fatal("Expected to be able to unreserve 1:", err) + } + if lru.TotalSize() != i-1 { + t.Fatalf("Expected total size %d, actual size %d", i-1, + lru.TotalSize()) + } + } + + if lru.TotalSize() != 0 { + t.Fatalf("Expected total size 0, actual size %d", + lru.TotalSize()) + } +} diff --git a/cache/httpproxy/httpproxy_test.go b/cache/httpproxy/httpproxy_test.go index 5ffee0779..23da27936 100644 --- a/cache/httpproxy/httpproxy_test.go +++ b/cache/httpproxy/httpproxy_test.go @@ -228,7 +228,7 @@ func TestEverything(t *testing.T) { diskCache = disk.New(cacheDir2, diskCacheSize, proxyCache) - _, numItems := diskCache.Stats() + _, _, numItems := diskCache.Stats() if numItems != 0 { t.Fatalf("Expected an empty disk cache, found %d items", numItems) } diff --git a/server/http.go b/server/http.go index 6a9c87d76..dc3e3f19e 100644 --- a/server/http.go +++ b/server/http.go @@ -37,11 +37,12 @@ type httpCache struct { } type statusPageData struct { - CurrSize int64 - MaxSize int64 - NumFiles int - ServerTime int64 - GitCommit string + CurrSize int64 + ReservedSize int64 + MaxSize int64 + NumFiles int + ServerTime int64 + GitCommit string } // NewHTTPCache returns a new instance of the cache. @@ -50,7 +51,7 @@ type statusPageData struct { // be reported. func NewHTTPCache(cache *disk.Cache, accessLogger cache.Logger, errorLogger cache.Logger, validateAC bool, commit string) HTTPCache { - _, numItems := cache.Stats() + _, _, numItems := cache.Stats() errorLogger.Printf("Loaded %d existing disk cache items.", numItems) @@ -348,17 +349,18 @@ func addWorkerMetadataHTTP(addr string, ct string, orig []byte) (data []byte, co func (h *httpCache) StatusPageHandler(w http.ResponseWriter, r *http.Request) { defer r.Body.Close() - currentSize, numItems := h.cache.Stats() + totalSize, reservedSize, numItems := h.cache.Stats() w.Header().Set("Content-Type", "application/json") enc := json.NewEncoder(w) enc.SetIndent("", " ") enc.Encode(statusPageData{ - MaxSize: h.cache.MaxSize(), - CurrSize: currentSize, - NumFiles: numItems, - ServerTime: time.Now().Unix(), - GitCommit: h.gitCommit, + MaxSize: h.cache.MaxSize(), + CurrSize: totalSize, + ReservedSize: reservedSize, + NumFiles: numItems, + ServerTime: time.Now().Unix(), + GitCommit: h.gitCommit, }) } diff --git a/server/http_test.go b/server/http_test.go index f4b6008cc..1e9b5a050 100644 --- a/server/http_test.go +++ b/server/http_test.go @@ -154,13 +154,15 @@ func TestUploadSameFileConcurrently(t *testing.T) { data, hash := testutils.RandomDataAndHash(1024) - c := disk.New(cacheDir, 1024, nil) + numWorkers := 100 + + c := disk.New(cacheDir, int64(len(data)*numWorkers), nil) h := NewHTTPCache(c, testutils.NewSilentLogger(), testutils.NewSilentLogger(), true, "") handler := http.HandlerFunc(h.CacheHandler) var wg sync.WaitGroup - wg.Add(100) - for i := 0; i < 100; i++ { + wg.Add(numWorkers) + for i := 0; i < numWorkers; i++ { go func() { defer wg.Done() diff --git a/utils/tempfile/BUILD.bazel b/utils/tempfile/BUILD.bazel new file mode 100644 index 000000000..136fe5ce0 --- /dev/null +++ b/utils/tempfile/BUILD.bazel @@ -0,0 +1,8 @@ +load("@io_bazel_rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["tempfile.go"], + importpath = "github.com/buchgr/bazel-remote/utils/tempfile", + visibility = ["//visibility:public"], +)