Repository: buchgr/bazel-remote
Pull request #323 (2020-07-27 22:52:04)

allow concurrent Puts and proxied Gets
At the moment, bazel-remote only allows a single concurrent Put for a given blob. If another Put starts, while one is ongoing, we discard the second copy and return quickly, but the blob is not yet available. This can cause trouble for the second client, which assumes that the blob is available as soon as its Put finishes. The same problem also exists for Get calls when using a proxy backend.

I guess this was not a problem when bazel-remote was first written and only supported the http cache protocol, because bazel serializes those requests more than it does with the newer gRPC protocol. With gRPC, a single bazel client is able to hit this problem.

This change refactors `disk.Cache` to allow concurrent Puts and proxied Gets, by reserving space in the LRU index and using pseudo-random tempfiles, While I was at it I extracted some common code into a couple of functions and reuse them in both Put and Get.

Hopefully this fixes #267 and #318.

mostynb: @buchgr: any thoughts on this change?

@ulrfa, @lizan: please let me know if you're able to test this with your builds.
buchgr: Thanks @mostynb . Taking a look!
mostynb: > Thanks @mostynb . Taking a look!

The diff is a bit confusing, it might be easier to look at the entire file:
https://github.com/buchgr/bazel-remote/blob/1621686f98d07baefc117f5e57a8c068a3b160c9/cache/disk/disk.go
lizan: @mostynb I was able to test at 7c1e9bd2305b521be0baecb86b0a4f696e26d44e and it fixed my issue.
lizan: @mostynb @buchgr thank you for working on this, any update? 
mostynb: > @mostynb @buchgr thank you for working on this, any update?

I will do another pass on this and try to land it next week.
dstranz: Hello @mostynb, 

We have similar issue when using bazel 3.4.1 and bazel-remote:master instance with s3 storage as remote-cache. 

I've just deploy bazel-remote from your branch on our server but it's not resolving this issue.

```
WARNING: Reading from Remote Cache:
BulkTransferException
WARNING: Writing to Remote Cache:
BulkTransferException
```
or
```
WARNING: Reading from Remote Cache:
ClosedChannelException
```

Let me know, if I could help you with testing your fix.
mostynb: @dstranz: are there any errors in bazel-remote's logs when you encounter this error? And do the bazel client errors mention any specific blob hashes that we can look for in bazel-remote's logs?
dstranz: @mostynb I will try to compare build output with bazel-remote logs tomorrow and let you know.
ulrfa: > @ulrfa, ... please let me know if you're able to test this with your builds.

I had a long summer vacation and is now trying to catch up. I will try to find time to review and test next week.

dstranz: @mostynb Looking into bazel-remote logs was a very good point. I figure out that our issue was related to s3 configuration - after fixing it everything works perfectly.
ulrfa: > I don't have access to a setup for high concurrency testing at the moment, So I have been mostly checking for correctness.

At Ericsson we deploy bazel-remote on 72 core machines. I plan to allocate some of them during the coming weekend, and evaluate if holding the mutex during syscall.Rename and os.Open, affects build performance or not, when there are many concurrent GET and PUT.

The previous conversation is easily lost in the collapsed code review for outdated code above. @mostynb, do you have more reflections about embedding both hash and size in the file path, to potentially reduce the need for holding mutex during the syscalls?

mostynb: > At Ericsson we deploy bazel-remote on 72 core machines. I plan to allocate some of them during the coming weekend, and evaluate if holding the mutex during syscall.Rename and os.Open, affects build performance or not, when there are many concurrent GET and PUT.

Thanks- that would be really useful.

> The previous conversation is easily lost in the collapsed code review for outdated code above. @mostynb, do you have more reflections about embedding both hash and size in the file path, to potentially reduce the need for holding mutex during the syscalls?

Unfortunately I think it might be racy, eg:

* abc-111 is in the index and on disk
* request 1: add abc-222 to disk
* request 1: add abc-222 to index
* request 2: add abc-111 to disk
* request 1: remove abc-111 from disk (no lock)
* request 2: add abc-111 to the index
* request 2: remove abc-222 from disk (no lock)

A more complicated version might work, where we write to files of the form <hash>.tmp<unique> then move them to <hash>.done<unique> and then insert them into the index (including a field with the <unique> string). If we're replacing an item in the index then we remove the old <hash>.done<otherunique> file from disk after releasing the lock. That way there is much lower chance of collision between racing requests.

But I think this change is large enough that we should land it in the more safe version (removing the files while holding the lock), and then follow up with an optimisation if it's too slow. Changing the on-disk format in this way probably requires a major version bump.
ulrfa: Perhaps a {hash}-{size} named file would not have to be removed by a later PUT of same hash but different size. Instead a new entry could be added to the LRU’s double linked list, and the original entry could still be in the double linked list and be evicted the normal way. That might involve having hash+size as key in the LRU, just as the gRPC protocol seems to do conceptually. And for HTTP with unspecified size, complement with an additional map in LRU with only hash as key, towards the most recent regardless of size. I think that could avoid the problematic race and still not require holding mutex during syscall.Rename.

However, I hope to get a chance to benchmark during the weekend. If no significant performance penalty, then we don’t need to bother about this and can keep it simple. :-)

ulrfa: I benchmarked three alternative implementations using mutexes in different ways:
 
- **mutex rename** = Almost identical to 983d785 from mostynb:concurrent_inserts
- **mutex minimal** = As mutex rename, but releases the mutex during syscall.Rename in commit.
- **mutex open + rename** = As mutex rename, but keeps the mutex locked also during os.Open in availableOrTryProxy.

Each implementation is benchmarked in two different scenarios. Each scenario is executed using up to 87 bazel clients started in parallell, for generating the load. Each client on separate machine. The bazel-remote server is deployed on a machine with 72 logical processors (2 x Xeon 6154) and 2 x 10 Gbit network interface (link aggregation).

### Scenario A: 0% Cache Hits

-	All clients doing same build, but with unique dummy –action_env to trigger cache misses.
-	Each bazel client send:
             ~7000 HTTP GET requests for non existing AC entries,
             ~30 000 HTTP PUT requests for uploading AC and CAS entries. 
             No HTTP HEAD. No gRPC.
-	Nothing evicted from cache during benchmark.
-	**Bazel-remote scales really well if not holding mutexes during system calls.**
-	**The graph indicates holding mutexes during syscall.Rename, starts to cause lock contention issues from 50 concurrent bazel clients. Increases build times 70% at 87 clients.**
-	The SSD seems to handle the load. Messured Write IOPS peek at 65 000 and sustained 35 000. (But it could be interesting to see how much time f.sync() takes...)

![bazelremote_mutex_graph_A](https://user-images.githubusercontent.com/59884749/90991680-5ebf8700-e5ab-11ea-84d5-dd36db475bc1.PNG)

### Scenario B: 100% Cache Hits

-	Each bazel client send:
	     ~36 000 HTTP GET requests downloading AC and CAS entries.
	     No HTTP PUT, No HTTP HEAD. No gRPC.
-	Almost no disk accesses, all data already in kernel’s RAM file system cache.
-	The 2 x 10 Gbit network interfaces become saturated from ~30 concurrent clients.


![bazelremote_mutex_graph_B](https://user-images.githubusercontent.com/59884749/90991675-59623c80-e5ab-11ea-9506-bd38893aa46b.PNG)
mostynb: Thanks for running these benchmarks. Here are some initial questions/notes:
* Did you hit eny errors during testing?
* In the hot-cache/100% cache hit scenario, it looks like "mutex rename" performs pretty well. And since the network connection is saturated we might need to wait for something like https://github.com/bazelbuild/remote-apis/issues/147 to improve on this significantly.
* There's clearly room for improvement over "mutex rename" in the cold-cache/0% cache hit scenario, but hopefully this isn't a common case to be in.
* Re `f.Sync()`, there's an old issue about this being slow on Mac hardware (#67). I had a prototype for working around this, but it has gone stale.
ulrfa: Thanks!

I believe the logic is correct, and did not notice any errors (related to this commit) during the load test.

If you decide to merge 983d785 now, despite the performance degradation, then it could perhaps be considered to embed both hash and file size in file path strings, the next time the on-disk format is changed for other reasons? (or address the degradation in other ways)

For the hot-cache/100% cache hit scenario, with saturated network interface, I agree that something like bazelbuild/remote-apis#147 and/or https://github.com/bazelbuild/bazel/issues/6862 would be needed.

mostynb: > Thanks!
> 
> I believe the logic is correct, and did not notice any errors (related to this commit) during the load test.
> 
> If you decide to merge [983d785](https://github.com/buchgr/bazel-remote/commit/983d785883a05e177f413aa93413f7c3b8e8a486) now, despite the performance degradation, then it could perhaps be considered to embed both hash and file size in file path strings, the next time the on-disk format is changed for other reasons? (or address the degradation in other ways)

This PR is pretty large already, I think it's best to land it with the potential performance hit but continue working on improvements to see if we can win some of that back. This is surely faster than a failed build.

We can change the on-disk format later if needed (and probably bump the version number to 2.0.0 in that case).

> For the hot-cache/100% cache hit scenario, with saturated network interface, I agree that something like [bazelbuild/remote-apis#147](https://github.com/bazelbuild/remote-apis/issues/147) and/or [bazelbuild/bazel#6862](https://github.com/bazelbuild/bazel/issues/6862) would be needed.

We can continue optimising after we get compression support (builds-without-the-bytes are already supported of course).

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/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/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()