Repository: buchgr/bazel-remote
Pull request #338 (2020-08-24 18:32:38)

explicitly close fds
The ReadCloser returned by disk.Cache.Get should be closed.

I had issues with exceeding max open file descriptors,
despite hard limit set to 100 000. Commit fe90bd1,
together with this one, resolved those issues.

mostynb: Nice one- I thought I caught this in the concurrent_inserts branch.

Patch:
diff --git a/README.md b/README.md
index 059dcb9..922a98b 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 f4006af..96a899d 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 40f9a6e..1805112 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/lru.go b/cache/disk/lru.go
index ab3f012..77a1079 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/server/grpc_ac.go b/server/grpc_ac.go
index 26c8430..7099e9c 100644
--- a/server/grpc_ac.go
+++ b/server/grpc_ac.go
@@ -61,6 +61,7 @@ func (s *grpcServer) GetActionResult(ctx context.Context,
 			return nil, status.Error(codes.NotFound,
 				fmt.Sprintf("%s not found in AC", req.ActionDigest.Hash))
 		}
+		defer rdr.Close()
 
 		acdata, err := ioutil.ReadAll(rdr)
 		if err != nil {
diff --git a/server/http.go b/server/http.go
index 6a9c87d..dc3e3f1 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/utils/tempfile/BUILD.bazel b/utils/tempfile/BUILD.bazel
new file mode 100644
index 0000000..136fe5c
--- /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"],
+)


Test patch:
diff --git a/cache/disk/disk_test.go b/cache/disk/disk_test.go
index bcb3a64..d4b3415 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_test.go b/cache/disk/lru_test.go
index c1ea363..2693916 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 5ffee07..23da279 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/grpc_bytestream.go b/server/grpc_bytestream.go
index 65b91d9..615c37c 100644
--- a/server/grpc_bytestream.go
+++ b/server/grpc_bytestream.go
@@ -105,6 +105,7 @@ func (s *grpcServer) Read(req *bytestream.ReadRequest,
 		s.accessLogger.Printf(msg)
 		return status.Error(codes.NotFound, msg)
 	}
+	defer rdr.Close()
 
 	if sizeBytes != size {
 		msg := fmt.Sprintf("Retrieved item had size %d expected %d",
diff --git a/server/http_test.go b/server/http_test.go
index f4b6008..1e9b5a0 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()