Support multiple non-overlapping write locks in SimpleCache

Issue: #5978
PiperOrigin-RevId: 313802629

RELEASENOTES.md

@@ -134,6 +134,8 @@
 *   Downloads and caching:
     *   Merge downloads in `SegmentDownloader` to improve overall download speed
         ([#5978](https://github.com/google/ExoPlayer/issues/5978)).
+    *   Support multiple non-overlapping write locks for the same key in
+        `SimpleCache`.
     *   Replace `CacheDataSinkFactory` and `CacheDataSourceFactory` with
         `CacheDataSink.Factory` and `CacheDataSource.Factory` respectively.
     *   Remove `DownloadConstructorHelper` and use `CacheDataSource.Factory`

library/core/src/main/java/com/google/android/exoplayer2/offline/SegmentDownloader.java

@@ -31,8 +31,10 @@ import com.google.android.exoplayer2.upstream.cache.CacheKeyFactory;
 import com.google.android.exoplayer2.upstream.cache.CacheWriter;
 import com.google.android.exoplayer2.upstream.cache.ContentMetadata;
 import com.google.android.exoplayer2.util.Assertions;
+import com.google.android.exoplayer2.util.Log;
 import com.google.android.exoplayer2.util.PriorityTaskManager;
 import com.google.android.exoplayer2.util.PriorityTaskManager.PriorityTooLowException;
+import com.google.android.exoplayer2.util.SystemClock;
 import com.google.android.exoplayer2.util.Util;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -218,17 +220,23 @@ public abstract class SegmentDownloader<M extends FilterableManifest<M>> impleme
     }
   }
 
+  long timer = 0;
+
   @Override
   public final void remove() {
     Cache cache = Assertions.checkNotNull(cacheDataSourceFactory.getCache());
     CacheKeyFactory cacheKeyFactory = cacheDataSourceFactory.getCacheKeyFactory();
     CacheDataSource dataSource = cacheDataSourceFactory.createDataSourceForRemovingDownload();
     try {
+      timer = SystemClock.DEFAULT.elapsedRealtime();
       M manifest = getManifest(dataSource, manifestDataSpec);
+      Log.e("XXX", "E1\t" + (SystemClock.DEFAULT.elapsedRealtime() - timer));
+      timer = SystemClock.DEFAULT.elapsedRealtime();
       List<Segment> segments = getSegments(dataSource, manifest, true);
       for (int i = 0; i < segments.size(); i++) {
         cache.removeResource(cacheKeyFactory.buildCacheKey(segments.get(i).dataSpec));
       }
+      Log.e("XXX", "E2\t" + (SystemClock.DEFAULT.elapsedRealtime() - timer));
     } catch (IOException e) {
       // Ignore exceptions when removing.
     } finally {

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/Cache.java

@@ -165,7 +165,7 @@ public interface Cache {
    * defines the file in which the data is stored. {@link CacheSpan#isCached} is true. The caller
    * may read from the cache file, but does not acquire any locks.
    *
-   * <p>If there is no cache entry overlapping {@code offset}, then the returned {@link CacheSpan}
+   * <p>If there is no cache entry overlapping {@code position}, then the returned {@link CacheSpan}
    * defines a hole in the cache starting at {@code position} into which the caller may write as it
    * obtains the data from some other source. The returned {@link CacheSpan} serves as a lock.
    * Whilst the caller holds the lock it may write data into the hole. It may split data into
@@ -177,31 +177,40 @@ public interface Cache {
    *
    * @param key The cache key of the resource.
    * @param position The starting position in the resource from which data is required.
+   * @param length The length of the data being requested, or {@link C#LENGTH_UNSET} if unbounded.
+   *     The length is ignored in the case of a cache hit. In the case of a cache miss, it defines
+   *     the maximum length of the hole {@link CacheSpan} that's returned. Cache implementations may
+   *     support parallel writes into non-overlapping holes, and so passing the actual required
+   *     length should be preferred to passing {@link C#LENGTH_UNSET} when possible.
    * @return The {@link CacheSpan}.
    * @throws InterruptedException If the thread was interrupted.
    * @throws CacheException If an error is encountered.
    */
   @WorkerThread
-  CacheSpan startReadWrite(String key, long position) throws InterruptedException, CacheException;
+  CacheSpan startReadWrite(String key, long position, long length)
+      throws InterruptedException, CacheException;
 
   /**
-   * Same as {@link #startReadWrite(String, long)}. However, if the cache entry is locked, then
+   * Same as {@link #startReadWrite(String, long, long)}. However, if the cache entry is locked,
-   * instead of blocking, this method will return null as the {@link CacheSpan}.
+   * then instead of blocking, this method will return null as the {@link CacheSpan}.
    *
    * <p>This method may be slow and shouldn't normally be called on the main thread.
    *
    * @param key The cache key of the resource.
    * @param position The starting position in the resource from which data is required.
+   * @param length The length of the data being requested, or {@link C#LENGTH_UNSET} if unbounded.
+   *     The length is ignored in the case of a cache hit. In the case of a cache miss, it defines
+   *     the range of data locked by the returned {@link CacheSpan}.
    * @return The {@link CacheSpan}. Or null if the cache entry is locked.
    * @throws CacheException If an error is encountered.
    */
   @WorkerThread
   @Nullable
-  CacheSpan startReadWriteNonBlocking(String key, long position) throws CacheException;
+  CacheSpan startReadWriteNonBlocking(String key, long position, long length) throws CacheException;
 
   /**
    * Obtains a cache file into which data can be written. Must only be called when holding a
-   * corresponding hole {@link CacheSpan} obtained from {@link #startReadWrite(String, long)}.
+   * corresponding hole {@link CacheSpan} obtained from {@link #startReadWrite(String, long, long)}.
    *
    * <p>This method may be slow and shouldn't normally be called on the main thread.
    *
@@ -217,7 +226,7 @@ public interface Cache {
 
   /**
    * Commits a file into the cache. Must only be called when holding a corresponding hole {@link
-   * CacheSpan} obtained from {@link #startReadWrite(String, long)}.
+   * CacheSpan} obtained from {@link #startReadWrite(String, long, long)}.
    *
    * <p>This method may be slow and shouldn't normally be called on the main thread.
    *
@@ -229,7 +238,7 @@ public interface Cache {
   void commitFile(File file, long length) throws CacheException;
 
   /**
-   * Releases a {@link CacheSpan} obtained from {@link #startReadWrite(String, long)} which
+   * Releases a {@link CacheSpan} obtained from {@link #startReadWrite(String, long, long)} which
    * corresponded to a hole in the cache.
    *
    * @param holeSpan The {@link CacheSpan} being released.

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/CacheDataSource.java

@@ -691,13 +691,13 @@ public final class CacheDataSource implements DataSource {
       nextSpan = null;
     } else if (blockOnCache) {
       try {
-        nextSpan = cache.startReadWrite(key, readPosition);
+        nextSpan = cache.startReadWrite(key, readPosition, bytesRemaining);
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         throw new InterruptedIOException();
       }
     } else {
-      nextSpan = cache.startReadWriteNonBlocking(key, readPosition);
+      nextSpan = cache.startReadWriteNonBlocking(key, readPosition, bytesRemaining);
     }
 
     DataSpec nextDataSpec;

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/CacheSpan.java

@@ -98,4 +98,8 @@ public class CacheSpan implements Comparable<CacheSpan> {
     return startOffsetDiff == 0 ? 0 : ((startOffsetDiff < 0) ? -1 : 1);
   }
 
+  @Override
+  public String toString() {
+    return "[" + position + ", " + length + "]";
+  }
 }

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/CachedContent.java

@@ -19,8 +19,10 @@ import static com.google.android.exoplayer2.util.Assertions.checkArgument;
 import static com.google.android.exoplayer2.util.Assertions.checkState;
 
 import androidx.annotation.Nullable;
+import com.google.android.exoplayer2.C;
 import com.google.android.exoplayer2.util.Log;
 import java.io.File;
+import java.util.ArrayList;
 import java.util.TreeSet;
 
 /** Defines the cached content for a single resource. */
@@ -34,10 +36,11 @@ import java.util.TreeSet;
   public final String key;
   /** The cached spans of this content. */
   private final TreeSet<SimpleCacheSpan> cachedSpans;
+  /** Currently locked ranges. */
+  private final ArrayList<Range> lockedRanges;
+
   /** Metadata values. */
   private DefaultContentMetadata metadata;
-  /** Whether the content is locked. */
-  private boolean locked;
 
   /**
    * Creates a CachedContent.
@@ -53,7 +56,8 @@ import java.util.TreeSet;
     this.id = id;
     this.key = key;
     this.metadata = metadata;
-    this.cachedSpans = new TreeSet<>();
+    cachedSpans = new TreeSet<>();
+    lockedRanges = new ArrayList<>();
   }
 
   /** Returns the metadata. */
@@ -72,14 +76,58 @@ import java.util.TreeSet;
     return !metadata.equals(oldMetadata);
   }
 
-  /** Returns whether the content is locked. */
+  /** Returns whether the entire resource is fully unlocked. */
-  public boolean isLocked() {
+  public boolean isFullyUnlocked() {
-    return locked;
+    return lockedRanges.isEmpty();
   }
 
-  /** Sets the locked state of the content. */
+  /**
-  public void setLocked(boolean locked) {
+   * Returns whether the specified range of the resource is fully locked by a single lock.
-    this.locked = locked;
+   *
+   * @param position The position of the range.
+   * @param length The length of the range, or {@link C#LENGTH_UNSET} if unbounded.
+   * @return Whether the range is fully locked by a single lock.
+   */
+  public boolean isFullyLocked(long position, long length) {
+    for (int i = 0; i < lockedRanges.size(); i++) {
+      if (lockedRanges.get(i).contains(position, length)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Attempts to lock the specified range of the resource.
+   *
+   * @param position The position of the range.
+   * @param length The length of the range, or {@link C#LENGTH_UNSET} if unbounded.
+   * @return Whether the range was successfully locked.
+   */
+  public boolean lockRange(long position, long length) {
+    for (int i = 0; i < lockedRanges.size(); i++) {
+      if (lockedRanges.get(i).intersects(position, length)) {
+        return false;
+      }
+    }
+    lockedRanges.add(new Range(position, length));
+    return true;
+  }
+
+  /**
+   * Unlocks the currently locked range starting at the specified position.
+   *
+   * @param position The starting position of the locked range.
+   * @throws IllegalStateException If there was no locked range starting at the specified position.
+   */
+  public void unlockRange(long position) {
+    for (int i = 0; i < lockedRanges.size(); i++) {
+      if (lockedRanges.get(i).position == position) {
+        lockedRanges.remove(i);
+        return;
+      }
+    }
+    throw new IllegalStateException();
   }
 
   /** Adds the given {@link SimpleCacheSpan} which contains a part of the content. */
@@ -93,18 +141,25 @@ import java.util.TreeSet;
   }
 
   /**
-   * Returns the span containing the position. If there isn't one, it returns a hole span
+   * Returns the cache span corresponding to the provided range. See {@link
-   * which defines the maximum extents of the hole in the cache.
+   * Cache#startReadWrite(String, long, long)} for detailed descriptions of the returned spans.
+   *
+   * @param position The position of the span being requested.
+   * @param length The length of the span, or {@link C#LENGTH_UNSET} if unbounded.
+   * @return The corresponding cache {@link SimpleCacheSpan}.
    */
-  public SimpleCacheSpan getSpan(long position) {
+  public SimpleCacheSpan getSpan(long position, long length) {
     SimpleCacheSpan lookupSpan = SimpleCacheSpan.createLookup(key, position);
     SimpleCacheSpan floorSpan = cachedSpans.floor(lookupSpan);
     if (floorSpan != null && floorSpan.position + floorSpan.length > position) {
       return floorSpan;
     }
     SimpleCacheSpan ceilSpan = cachedSpans.ceiling(lookupSpan);
-    return ceilSpan == null ? SimpleCacheSpan.createOpenHole(key, position)
+    if (ceilSpan != null) {
-        : SimpleCacheSpan.createClosedHole(key, position, ceilSpan.position - position);
+      long holeLength = ceilSpan.position - position;
+      length = length == C.LENGTH_UNSET ? holeLength : Math.min(holeLength, length);
+    }
+    return SimpleCacheSpan.createHole(key, position, length);
   }
 
   /**
@@ -121,7 +176,7 @@ import java.util.TreeSet;
   public long getCachedBytesLength(long position, long length) {
     checkArgument(position >= 0);
     checkArgument(length >= 0);
-    SimpleCacheSpan span = getSpan(position);
+    SimpleCacheSpan span = getSpan(position, length);
     if (span.isHoleSpan()) {
       // We don't have a span covering the start of the queried region.
       return -Math.min(span.isOpenEnded() ? Long.MAX_VALUE : span.length, length);
@@ -215,4 +270,51 @@ import java.util.TreeSet;
         && cachedSpans.equals(that.cachedSpans)
         && metadata.equals(that.metadata);
   }
+
+  private static final class Range {
+
+    /** The starting position of the range. */
+    public final long position;
+    /** The length of the range, or {@link C#LENGTH_UNSET} if unbounded. */
+    public final long length;
+
+    public Range(long position, long length) {
+      this.position = position;
+      this.length = length;
+    }
+
+    /**
+     * Returns whether this range fully contains the range specified by {@code otherPosition} and
+     * {@code otherLength}.
+     *
+     * @param otherPosition The position of the range to check.
+     * @param otherLength The length of the range to check, or {@link C#LENGTH_UNSET} if unbounded.
+     * @return Whether this range fully contains the specified range.
+     */
+    public boolean contains(long otherPosition, long otherLength) {
+      if (length == C.LENGTH_UNSET) {
+        return otherPosition >= position;
+      } else if (otherLength == C.LENGTH_UNSET) {
+        return false;
+      } else {
+        return position <= otherPosition && (otherPosition + otherLength) <= (position + length);
+      }
+    }
+
+    /**
+     * Returns whether this range intersects with the range specified by {@code otherPosition} and
+     * {@code otherLength}.
+     *
+     * @param otherPosition The position of the range to check.
+     * @param otherLength The length of the range to check, or {@link C#LENGTH_UNSET} if unbounded.
+     * @return Whether this range intersects with the specified range.
+     */
+    public boolean intersects(long otherPosition, long otherLength) {
+      if (position <= otherPosition) {
+        return length == C.LENGTH_UNSET || position + length > otherPosition;
+      } else {
+        return otherLength == C.LENGTH_UNSET || otherPosition + otherLength > position;
+      }
+    }
+  }
 }

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/CachedContentIndex.java

@@ -273,7 +273,7 @@ import org.checkerframework.checker.nullness.compatqual.NullableType;
    */
   public void maybeRemove(String key) {
     @Nullable CachedContent cachedContent = keyToContent.get(key);
-    if (cachedContent != null && cachedContent.isEmpty() && !cachedContent.isLocked()) {
+    if (cachedContent != null && cachedContent.isEmpty() && cachedContent.isFullyUnlocked()) {
       keyToContent.remove(key);
       int id = cachedContent.id;
       boolean neverStored = newIds.get(id);

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/SimpleCache.java

@@ -353,13 +353,13 @@ public final class SimpleCache implements Cache {
   }
 
   @Override
-  public synchronized CacheSpan startReadWrite(String key, long position)
+  public synchronized CacheSpan startReadWrite(String key, long position, long length)
       throws InterruptedException, CacheException {
     Assertions.checkState(!released);
     checkInitialization();
 
     while (true) {
-      CacheSpan span = startReadWriteNonBlocking(key, position);
+      CacheSpan span = startReadWriteNonBlocking(key, position, length);
       if (span != null) {
         return span;
       } else {
@@ -375,12 +375,12 @@ public final class SimpleCache implements Cache {
 
   @Override
   @Nullable
-  public synchronized CacheSpan startReadWriteNonBlocking(String key, long position)
+  public synchronized CacheSpan startReadWriteNonBlocking(String key, long position, long length)
       throws CacheException {
     Assertions.checkState(!released);
     checkInitialization();
 
-    SimpleCacheSpan span = getSpan(key, position);
+    SimpleCacheSpan span = getSpan(key, position, length);
 
     if (span.isCached) {
       // Read case.
@@ -388,9 +388,8 @@ public final class SimpleCache implements Cache {
     }
 
     CachedContent cachedContent = contentIndex.getOrAdd(key);
-    if (!cachedContent.isLocked()) {
+    if (cachedContent.lockRange(position, span.length)) {
       // Write case.
-      cachedContent.setLocked(true);
       return span;
     }
 
@@ -405,7 +404,7 @@ public final class SimpleCache implements Cache {
 
     CachedContent cachedContent = contentIndex.get(key);
     Assertions.checkNotNull(cachedContent);
-    Assertions.checkState(cachedContent.isLocked());
+    Assertions.checkState(cachedContent.isFullyLocked(position, length));
     if (!cacheDir.exists()) {
       // For some reason the cache directory doesn't exist. Make a best effort to create it.
       cacheDir.mkdirs();
@@ -435,7 +434,7 @@ public final class SimpleCache implements Cache {
     SimpleCacheSpan span =
         Assertions.checkNotNull(SimpleCacheSpan.createCacheEntry(file, length, contentIndex));
     CachedContent cachedContent = Assertions.checkNotNull(contentIndex.get(span.key));
-    Assertions.checkState(cachedContent.isLocked());
+    Assertions.checkState(cachedContent.isFullyLocked(span.position, span.length));
 
     // Check if the span conflicts with the set content length
     long contentLength = ContentMetadata.getContentLength(cachedContent.getMetadata());
@@ -464,8 +463,7 @@ public final class SimpleCache implements Cache {
   public synchronized void releaseHoleSpan(CacheSpan holeSpan) {
     Assertions.checkState(!released);
     CachedContent cachedContent = Assertions.checkNotNull(contentIndex.get(holeSpan.key));
-    Assertions.checkState(cachedContent.isLocked());
+    cachedContent.unlockRange(holeSpan.position);
-    cachedContent.setLocked(false);
     contentIndex.maybeRemove(cachedContent.key);
     notifyAll();
   }
@@ -688,23 +686,21 @@ public final class SimpleCache implements Cache {
   }
 
   /**
-   * Returns the cache span corresponding to the provided lookup span.
+   * Returns the cache span corresponding to the provided key and range. See {@link
-   *
+   * Cache#startReadWrite(String, long, long)} for detailed descriptions of the returned spans.
-   * <p>If the lookup position is contained by an existing entry in the cache, then the returned
-   * span defines the file in which the data is stored. If the lookup position is not contained by
-   * an existing entry, then the returned span defines the maximum extents of the hole in the cache.
    *
    * @param key The key of the span being requested.
    * @param position The position of the span being requested.
+   * @param length The length of the span, or {@link C#LENGTH_UNSET} if unbounded.
    * @return The corresponding cache {@link SimpleCacheSpan}.
    */
-  private SimpleCacheSpan getSpan(String key, long position) {
+  private SimpleCacheSpan getSpan(String key, long position, long length) {
     @Nullable CachedContent cachedContent = contentIndex.get(key);
     if (cachedContent == null) {
-      return SimpleCacheSpan.createOpenHole(key, position);
+      return SimpleCacheSpan.createHole(key, position, length);
     }
     while (true) {
-      SimpleCacheSpan span = cachedContent.getSpan(position);
+      SimpleCacheSpan span = cachedContent.getSpan(position, length);
       if (span.isCached && span.file.length() != span.length) {
         // The file has been modified or deleted underneath us. It's likely that other files will
         // have been modified too, so scan the whole in-memory representation.

library/core/src/main/java/com/google/android/exoplayer2/upstream/cache/SimpleCacheSpan.java

@@ -54,7 +54,7 @@ import java.util.regex.Pattern;
    * Creates a lookup span.
    *
    * @param key The cache key of the resource.
-   * @param position The position of the {@link CacheSpan} in the resource.
+   * @param position The position of the span in the resource.
    * @return The span.
    */
   public static SimpleCacheSpan createLookup(String key, long position) {
@@ -62,25 +62,14 @@ import java.util.regex.Pattern;
   }
 
   /**
-   * Creates an open hole span.
+   * Creates a hole span.
    *
    * @param key The cache key of the resource.
-   * @param position The position of the {@link CacheSpan} in the resource.
+   * @param position The position of the span in the resource.
-   * @return The span.
+   * @param length The length of the span, or {@link C#LENGTH_UNSET} if unbounded.
+   * @return The hole span.
    */
-  public static SimpleCacheSpan createOpenHole(String key, long position) {
+  public static SimpleCacheSpan createHole(String key, long position, long length) {
-    return new SimpleCacheSpan(key, position, C.LENGTH_UNSET, C.TIME_UNSET, null);
-  }
-
-  /**
-   * Creates a closed hole span.
-   *
-   * @param key The cache key of the resource.
-   * @param position The position of the {@link CacheSpan} in the resource.
-   * @param length The length of the {@link CacheSpan}.
-   * @return The span.
-   */
-  public static SimpleCacheSpan createClosedHole(String key, long position, long length) {
     return new SimpleCacheSpan(key, position, length, C.TIME_UNSET, null);
   }
 
@@ -191,12 +180,11 @@ import java.util.regex.Pattern;
 
   /**
    * @param key The cache key of the resource.
-   * @param position The position of the {@link CacheSpan} in the resource.
+   * @param position The position of the span in the resource.
-   * @param length The length of the {@link CacheSpan}, or {@link C#LENGTH_UNSET} if this is an
+   * @param length The length of the span, or {@link C#LENGTH_UNSET} if this is an open-ended hole.
-   *     open-ended hole.
    * @param lastTouchTimestamp The last touch timestamp, or {@link C#TIME_UNSET} if {@link
    *     #isCached} is false.
-   * @param file The file corresponding to this {@link CacheSpan}, or null if it's a hole.
+   * @param file The file corresponding to this span, or null if it's a hole.
    */
   private SimpleCacheSpan(
       String key, long position, long length, long lastTouchTimestamp, @Nullable File file) {

library/core/src/test/java/com/google/android/exoplayer2/upstream/cache/CacheDataSourceTest.java

@@ -384,7 +384,7 @@ public final class CacheDataSourceTest {
         .appendReadData(1);
 
     // Lock the content on the cache.
-    CacheSpan cacheSpan = cache.startReadWriteNonBlocking(defaultCacheKey, 0);
+    CacheSpan cacheSpan = cache.startReadWriteNonBlocking(defaultCacheKey, 0, C.LENGTH_UNSET);
     assertThat(cacheSpan).isNotNull();
     assertThat(cacheSpan.isHoleSpan()).isTrue();
 

library/core/src/test/java/com/google/android/exoplayer2/upstream/cache/CachedContentIndexTest.java

@@ -301,7 +301,7 @@ public class CachedContentIndexTest {
   public void cantRemoveLockedCachedContent() {
     CachedContentIndex index = newInstance();
     CachedContent cachedContent = index.getOrAdd("key1");
-    cachedContent.setLocked(true);
+    cachedContent.lockRange(0, 1);
 
     index.maybeRemove(cachedContent.key);
 

library/core/src/test/java/com/google/android/exoplayer2/upstream/cache/SimpleCacheTest.java

@@ -100,7 +100,7 @@ public class SimpleCacheTest {
     SimpleCache simpleCache = getSimpleCache();
 
     // Write some data and metadata to the cache.
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 15);
     simpleCache.releaseHoleSpan(holeSpan);
     ContentMetadataMutations mutations = new ContentMetadataMutations();
@@ -112,7 +112,7 @@ public class SimpleCacheTest {
     simpleCache = getSimpleCache();
 
     // Read the cached data and metadata back.
-    CacheSpan fileSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan fileSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     assertCachedDataReadCorrect(fileSpan);
     assertThat(ContentMetadata.getRedirectedUri(simpleCache.getContentMetadata(KEY_1)))
         .isEqualTo(Uri.parse("https://redirect.google.com"));
@@ -130,7 +130,7 @@ public class SimpleCacheTest {
   public void newInstance_withExistingCacheDirectory_resolvesInconsistentState() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
 
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 15);
     simpleCache.releaseHoleSpan(holeSpan);
     simpleCache.removeSpan(simpleCache.getCachedSpans(KEY_1).first());
@@ -151,7 +151,7 @@ public class SimpleCacheTest {
   @Test
   public void newInstance_withEncryptedIndex() throws Exception {
     SimpleCache simpleCache = getEncryptedSimpleCache(ENCRYPTED_INDEX_KEY);
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 15);
     simpleCache.releaseHoleSpan(holeSpan);
     simpleCache.release();
@@ -160,7 +160,7 @@ public class SimpleCacheTest {
     simpleCache = getEncryptedSimpleCache(ENCRYPTED_INDEX_KEY);
 
     // Read the cached data back.
-    CacheSpan fileSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan fileSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     assertCachedDataReadCorrect(fileSpan);
   }
 
@@ -169,7 +169,7 @@ public class SimpleCacheTest {
     SimpleCache simpleCache = getEncryptedSimpleCache(ENCRYPTED_INDEX_KEY);
 
     // Write data.
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 15);
     simpleCache.releaseHoleSpan(holeSpan);
     simpleCache.release();
@@ -187,7 +187,7 @@ public class SimpleCacheTest {
     SimpleCache simpleCache = getEncryptedSimpleCache(ENCRYPTED_INDEX_KEY);
 
     // Write data.
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 15);
     simpleCache.releaseHoleSpan(holeSpan);
     simpleCache.release();
@@ -204,59 +204,179 @@ public class SimpleCacheTest {
   public void write_oneLock_oneFile_thenRead() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
 
-    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
-    assertThat(holeSpan1.isCached).isFalse();
+    assertThat(holeSpan.isCached).isFalse();
-    assertThat(holeSpan1.isOpenEnded()).isTrue();
+    assertThat(holeSpan.isOpenEnded()).isTrue();
     addCache(simpleCache, KEY_1, 0, 15);
 
-    CacheSpan readSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan readSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     assertThat(readSpan.position).isEqualTo(0);
     assertThat(readSpan.length).isEqualTo(15);
     assertCachedDataReadCorrect(readSpan);
     assertThat(simpleCache.getCacheSpace()).isEqualTo(15);
+
+    simpleCache.releaseHoleSpan(holeSpan);
   }
 
   @Test
   public void write_oneLock_twoFiles_thenRead() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
 
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 7);
     addCache(simpleCache, KEY_1, 7, 8);
 
-    CacheSpan readSpan1 = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan readSpan1 = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     assertThat(readSpan1.position).isEqualTo(0);
     assertThat(readSpan1.length).isEqualTo(7);
     assertCachedDataReadCorrect(readSpan1);
-    CacheSpan readSpan2 = simpleCache.startReadWrite(KEY_1, 7);
+    CacheSpan readSpan2 = simpleCache.startReadWrite(KEY_1, 7, LENGTH_UNSET);
     assertThat(readSpan2.position).isEqualTo(7);
     assertThat(readSpan2.length).isEqualTo(8);
     assertCachedDataReadCorrect(readSpan2);
     assertThat(simpleCache.getCacheSpace()).isEqualTo(15);
+
+    simpleCache.releaseHoleSpan(holeSpan);
+  }
+
+  @Test
+  public void write_twoLocks_twoFiles_thenRead() throws Exception {
+    SimpleCache simpleCache = getSimpleCache();
+
+    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 0, 7);
+    CacheSpan holeSpan2 = simpleCache.startReadWrite(KEY_1, 7, 8);
+
+    addCache(simpleCache, KEY_1, 0, 7);
+    addCache(simpleCache, KEY_1, 7, 8);
+
+    CacheSpan readSpan1 = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
+    assertThat(readSpan1.position).isEqualTo(0);
+    assertThat(readSpan1.length).isEqualTo(7);
+    assertCachedDataReadCorrect(readSpan1);
+    CacheSpan readSpan2 = simpleCache.startReadWrite(KEY_1, 7, LENGTH_UNSET);
+    assertThat(readSpan2.position).isEqualTo(7);
+    assertThat(readSpan2.length).isEqualTo(8);
+    assertCachedDataReadCorrect(readSpan2);
+    assertThat(simpleCache.getCacheSpace()).isEqualTo(15);
+
+    simpleCache.releaseHoleSpan(holeSpan1);
+    simpleCache.releaseHoleSpan(holeSpan2);
   }
 
   @Test
   public void write_differentKeyLocked_thenRead() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 50);
+    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
 
-    CacheSpan holeSpan2 = simpleCache.startReadWrite(KEY_2, 50);
+    CacheSpan holeSpan2 = simpleCache.startReadWrite(KEY_2, 0, LENGTH_UNSET);
-    assertThat(holeSpan1.isCached).isFalse();
+    assertThat(holeSpan2.isCached).isFalse();
-    assertThat(holeSpan1.isOpenEnded()).isTrue();
+    assertThat(holeSpan2.isOpenEnded()).isTrue();
     addCache(simpleCache, KEY_2, 0, 15);
 
-    CacheSpan readSpan = simpleCache.startReadWrite(KEY_2, 0);
+    CacheSpan readSpan = simpleCache.startReadWrite(KEY_2, 0, LENGTH_UNSET);
     assertThat(readSpan.length).isEqualTo(15);
     assertCachedDataReadCorrect(readSpan);
     assertThat(simpleCache.getCacheSpace()).isEqualTo(15);
+
+    simpleCache.releaseHoleSpan(holeSpan1);
+    simpleCache.releaseHoleSpan(holeSpan2);
   }
 
   @Test
-  public void write_sameKeyLocked_fails() throws Exception {
+  public void write_oneLock_fileExceedsLock_fails() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan cacheSpan1 = simpleCache.startReadWrite(KEY_1, 50);
 
-    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 25)).isNull();
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, 10);
+
+    assertThrows(IllegalStateException.class, () -> addCache(simpleCache, KEY_1, 0, 11));
+
+    simpleCache.releaseHoleSpan(holeSpan);
+  }
+
+  @Test
+  public void write_twoLocks_oneFileSpanningBothLocks_fails() throws Exception {
+    SimpleCache simpleCache = getSimpleCache();
+
+    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 0, 7);
+    CacheSpan holeSpan2 = simpleCache.startReadWrite(KEY_1, 7, 8);
+
+    assertThrows(IllegalStateException.class, () -> addCache(simpleCache, KEY_1, 0, 15));
+
+    simpleCache.releaseHoleSpan(holeSpan1);
+    simpleCache.releaseHoleSpan(holeSpan2);
+  }
+
+  @Test
+  public void write_unboundedRangeLocked_lockingOverlappingRange_fails() throws Exception {
+    SimpleCache simpleCache = getSimpleCache();
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 50, LENGTH_UNSET);
+
+    // Overlapping cannot be locked.
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 49, 2)).isNull();
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 99, 2)).isNull();
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 0, LENGTH_UNSET)).isNull();
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 9, LENGTH_UNSET)).isNull();
+
+    simpleCache.releaseHoleSpan(holeSpan);
+  }
+
+  @Test
+  public void write_unboundedRangeLocked_lockingNonOverlappingRange_succeeds() throws Exception {
+    SimpleCache simpleCache = getSimpleCache();
+    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 50, LENGTH_UNSET);
+
+    // Non-overlapping range can be locked.
+    CacheSpan holeSpan2 = simpleCache.startReadWrite(KEY_1, 0, 50);
+    assertThat(holeSpan2.isCached).isFalse();
+    assertThat(holeSpan2.position).isEqualTo(0);
+    assertThat(holeSpan2.length).isEqualTo(50);
+
+    simpleCache.releaseHoleSpan(holeSpan1);
+    simpleCache.releaseHoleSpan(holeSpan2);
+  }
+
+  @Test
+  public void write_boundedRangeLocked_lockingOverlappingRange_fails() throws Exception {
+    SimpleCache simpleCache = getSimpleCache();
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 50, 50);
+
+    // Overlapping cannot be locked.
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 49, 2)).isNull();
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 99, 2)).isNull();
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 0, LENGTH_UNSET)).isNull();
+    assertThat(simpleCache.startReadWriteNonBlocking(KEY_1, 99, LENGTH_UNSET)).isNull();
+
+    simpleCache.releaseHoleSpan(holeSpan);
+  }
+
+  @Test
+  public void write_boundedRangeLocked_lockingNonOverlappingRange_succeeds() throws Exception {
+    SimpleCache simpleCache = getSimpleCache();
+
+    CacheSpan holeSpan1 = simpleCache.startReadWrite(KEY_1, 50, 50);
+    assertThat(holeSpan1.isCached).isFalse();
+    assertThat(holeSpan1.length).isEqualTo(50);
+
+    // Non-overlapping range can be locked.
+    CacheSpan holeSpan2 = simpleCache.startReadWriteNonBlocking(KEY_1, 49, 1);
+    assertThat(holeSpan2.isCached).isFalse();
+    assertThat(holeSpan2.position).isEqualTo(49);
+    assertThat(holeSpan2.length).isEqualTo(1);
+    simpleCache.releaseHoleSpan(holeSpan2);
+
+    CacheSpan holeSpan3 = simpleCache.startReadWriteNonBlocking(KEY_1, 100, 1);
+    assertThat(holeSpan3.isCached).isFalse();
+    assertThat(holeSpan3.position).isEqualTo(100);
+    assertThat(holeSpan3.length).isEqualTo(1);
+    simpleCache.releaseHoleSpan(holeSpan3);
+
+    CacheSpan holeSpan4 = simpleCache.startReadWriteNonBlocking(KEY_1, 100, LENGTH_UNSET);
+    assertThat(holeSpan4.isCached).isFalse();
+    assertThat(holeSpan4.position).isEqualTo(100);
+    assertThat(holeSpan4.isOpenEnded()).isTrue();
+    simpleCache.releaseHoleSpan(holeSpan4);
+
+    simpleCache.releaseHoleSpan(holeSpan1);
   }
 
   @Test
@@ -275,11 +395,11 @@ public class SimpleCacheTest {
   @Test
   public void removeSpans_removesSpansWithSameKey() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 10);
     addCache(simpleCache, KEY_1, 20, 10);
     simpleCache.releaseHoleSpan(holeSpan);
-    holeSpan = simpleCache.startReadWrite(KEY_2, 20);
+    holeSpan = simpleCache.startReadWrite(KEY_2, 20, LENGTH_UNSET);
     addCache(simpleCache, KEY_2, 20, 10);
     simpleCache.releaseHoleSpan(holeSpan);
 
@@ -309,7 +429,7 @@ public class SimpleCacheTest {
   @Test
   public void getCachedLength_returnsNegativeHoleLength() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, /* position= */ 50, /* length= */ 50);
     simpleCache.releaseHoleSpan(holeSpan);
 
@@ -330,7 +450,7 @@ public class SimpleCacheTest {
   @Test
   public void getCachedLength_returnsCachedLength() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, /* position= */ 0, /* length= */ 50);
     simpleCache.releaseHoleSpan(holeSpan);
 
@@ -353,7 +473,7 @@ public class SimpleCacheTest {
   @Test
   public void getCachedLength_withMultipleAdjacentSpans_returnsCachedLength() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, /* position= */ 0, /* length= */ 25);
     addCache(simpleCache, KEY_1, /* position= */ 25, /* length= */ 25);
     simpleCache.releaseHoleSpan(holeSpan);
@@ -377,7 +497,7 @@ public class SimpleCacheTest {
   @Test
   public void getCachedLength_withMultipleNonAdjacentSpans_returnsCachedLength() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, /* position= */ 0, /* length= */ 10);
     addCache(simpleCache, KEY_1, /* position= */ 15, /* length= */ 35);
     simpleCache.releaseHoleSpan(holeSpan);
@@ -419,7 +539,7 @@ public class SimpleCacheTest {
   @Test
   public void getCachedBytes_withMultipleAdjacentSpans_returnsCachedBytes() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, /* position= */ 0, /* length= */ 25);
     addCache(simpleCache, KEY_1, /* position= */ 25, /* length= */ 25);
     simpleCache.releaseHoleSpan(holeSpan);
@@ -443,7 +563,7 @@ public class SimpleCacheTest {
   @Test
   public void getCachedBytes_withMultipleNonAdjacentSpans_returnsCachedBytes() throws Exception {
     SimpleCache simpleCache = getSimpleCache();
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, /* position= */ 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, /* position= */ 0, /* length= */ 10);
     addCache(simpleCache, KEY_1, /* position= */ 15, /* length= */ 35);
     simpleCache.releaseHoleSpan(holeSpan);
@@ -474,7 +594,7 @@ public class SimpleCacheTest {
             cacheDir, new LeastRecentlyUsedCacheEvictor(20), contentIndex, /* fileIndex= */ null);
 
     // Add some content.
-    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0);
+    CacheSpan holeSpan = simpleCache.startReadWrite(KEY_1, 0, LENGTH_UNSET);
     addCache(simpleCache, KEY_1, 0, 15);
 
     // Make index.store() throw exception from now on.
@@ -502,7 +622,8 @@ public class SimpleCacheTest {
     SimpleCache simpleCache = new SimpleCache(cacheDir, new NoOpCacheEvictor());
     simpleCache.release();
     assertThrows(
-        IllegalStateException.class, () -> simpleCache.startReadWriteNonBlocking(KEY_1, 0));
+        IllegalStateException.class,
+        () -> simpleCache.startReadWriteNonBlocking(KEY_1, 0, LENGTH_UNSET));
   }
 
   private SimpleCache getSimpleCache() {