Move caching of the size of a directory to StoreDirectory.

server/src/main/java/org/elasticsearch/common/util/SingleObjectCache.java

@@ -64,6 +64,11 @@ public T getOrRefresh() {
         return cached;
     }
 
+    /** Return the potentially stale cached entry. */
+    protected final T getNoRefresh() {
+        return cached;
+    }
+
     /**
      * Returns a new instance to cache
      */

server/src/main/java/org/elasticsearch/index/store/ByteSizeCachingDirectory.java

@@ -0,0 +1,144 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.store;
+
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FilterDirectory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexOutput;
+import org.elasticsearch.common.lucene.store.FilterIndexOutput;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.SingleObjectCache;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.NoSuchFileException;
+import java.util.concurrent.atomic.AtomicLong;
+
+final class ByteSizeCachingDirectory extends FilterDirectory {
+
+    private static class SizeAndModCount {
+        final long size;
+        final long modCount;
+
+        SizeAndModCount(long length, long modCount) {
+            this.size = length;
+            this.modCount = modCount;
+        }
+    }
+
+    private static long estimateSize(Directory directory) throws IOException {
+        long estimatedSize = 0;
+        String[] files = directory.listAll();
+        for (String file : files) {
+            try {
+                estimatedSize += directory.fileLength(file);
+            } catch (NoSuchFileException | FileNotFoundException | AccessDeniedException e) {
+                // ignore, the file is not there no more; on Windows, if one thread concurrently deletes a file while
+                // calling Files.size, you can also sometimes hit AccessDeniedException
+            }
+        }
+        return estimatedSize;
+    }
+
+    private final AtomicLong modCount = new AtomicLong();
+    private final SingleObjectCache<SizeAndModCount> size;
+
+    ByteSizeCachingDirectory(Directory in, TimeValue refreshInterval) {
+        super(in);
+        size = new SingleObjectCache<SizeAndModCount>(refreshInterval, new SizeAndModCount(0L, -1L)) {
+            @Override
+            protected SizeAndModCount refresh() {
+                // Compute modCount first so that updates that happen while the size
+                // is being computed invalidate the length
+                final long modCount = ByteSizeCachingDirectory.this.modCount.get();
+                final long size;
+                try {
+                    size = estimateSize(getDelegate());
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+                return new SizeAndModCount(size, modCount);
+            }
+
+            @Override
+            protected boolean needsRefresh() {
+                if (getNoRefresh().modCount == modCount.get()) {
+                    // no updates to the directory since the last refresh
+                    return false;
+                }
+                return super.needsRefresh();
+            }
+        };
+    }
+
+    /** Return the cumulative size of all files in this directory. */
+    long estimateSize() throws IOException {
+        try {
+            return size.getOrRefresh().size;
+        } catch (UncheckedIOException e) {
+            // we wrapped in the cache and unwrap here
+            throw e.getCause();
+        }
+    }
+
+    @Override
+    public IndexOutput createOutput(String name, IOContext context) throws IOException {
+        return wrapIndexOutput(super.createOutput(name, context));
+    }
+
+    @Override
+    public IndexOutput createTempOutput(String prefix, String suffix, IOContext context) throws IOException {
+        return wrapIndexOutput(super.createTempOutput(prefix, suffix, context));
+    }
+
+    private IndexOutput wrapIndexOutput(IndexOutput out) {
+        return new FilterIndexOutput(out.toString(), out) {
+            @Override
+            public void writeBytes(byte[] b, int length) throws IOException {
+                super.writeBytes(b, length);
+                modCount.incrementAndGet();
+            }
+
+            @Override
+            public void writeByte(byte b) throws IOException {
+                super.writeByte(b);
+                modCount.incrementAndGet();
+            }
+
+            @Override
+            public void close() throws IOException {
+                // Close might cause some data to be flushed from in-memory buffers, so
+                // increment the modification counter too.
+                super.close();
+                modCount.incrementAndGet();
+            }
+        };
+    }
+
+    @Override
+    public void deleteFile(String name) throws IOException {
+        super.deleteFile(name);
+        modCount.incrementAndGet();
+    }
+
+}

server/src/main/java/org/elasticsearch/index/store/Store.java

@@ -50,7 +50,6 @@
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefBuilder;
 import org.apache.lucene.util.Version;
-import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -67,7 +66,6 @@
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.SingleObjectCache;
 import org.elasticsearch.common.util.concurrent.AbstractRefCounted;
 import org.elasticsearch.common.util.concurrent.RefCounted;
 import org.elasticsearch.common.util.iterable.Iterables;
@@ -91,7 +89,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintStream;
-import java.nio.file.AccessDeniedException;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.util.ArrayList;
@@ -146,7 +143,6 @@ public class Store extends AbstractIndexShardComponent implements Closeable, Ref
     private final ReentrantReadWriteLock metadataLock = new ReentrantReadWriteLock();
     private final ShardLock shardLock;
     private final OnClose onClose;
-    private final SingleObjectCache<StoreStats> statsCache;
 
     private final AbstractRefCounted refCounter = new AbstractRefCounted("store") {
         @Override
@@ -164,12 +160,13 @@ public Store(ShardId shardId, IndexSettings indexSettings, DirectoryService dire
                  OnClose onClose) throws IOException {
         super(shardId, indexSettings);
         final Settings settings = indexSettings.getSettings();
-        this.directory = new StoreDirectory(directoryService.newDirectory(), Loggers.getLogger("index.store.deletes", settings, shardId));
-        this.shardLock = shardLock;
-        this.onClose = onClose;
+        Directory dir = directoryService.newDirectory();
         final TimeValue refreshInterval = indexSettings.getValue(INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING);
-        this.statsCache = new StoreStatsCache(refreshInterval, directory);
         logger.debug("store stats are refreshed with refresh_interval [{}]", refreshInterval);
+        ByteSizeCachingDirectory sizeCachingDir = new ByteSizeCachingDirectory(dir, refreshInterval);
+        this.directory = new StoreDirectory(sizeCachingDir, Loggers.getLogger("index.store.deletes", settings, shardId));
+        this.shardLock = shardLock;
+        this.onClose = onClose;
 
         assert onClose != null;
         assert shardLock != null;
@@ -377,7 +374,7 @@ public void exorciseIndex(CheckIndex.Status status) throws IOException {
 
     public StoreStats stats() throws IOException {
         ensureOpen();
-        return statsCache.getOrRefresh();
+        return new StoreStats(directory.estimateSize());
     }
 
     /**
@@ -731,11 +728,16 @@ static final class StoreDirectory extends FilterDirectory {
 
         private final Logger deletesLogger;
 
-        StoreDirectory(Directory delegateDirectory, Logger deletesLogger) {
+        StoreDirectory(ByteSizeCachingDirectory delegateDirectory, Logger deletesLogger) {
             super(delegateDirectory);
             this.deletesLogger = deletesLogger;
         }
 
+        /** Estimate the cumulative size of all files in this directory in bytes. */
+        long estimateSize() throws IOException {
+            return ((ByteSizeCachingDirectory) getDelegate()).estimateSize();
+        }
+
         @Override
         public void close() {
             assert false : "Nobody should close this directory except of the Store itself";
@@ -1439,38 +1441,6 @@ public void accept(ShardLock Lock) {
         };
     }
 
-    private static class StoreStatsCache extends SingleObjectCache<StoreStats> {
-        private final Directory directory;
-
-        StoreStatsCache(TimeValue refreshInterval, Directory directory) throws IOException {
-            super(refreshInterval, new StoreStats(estimateSize(directory)));
-            this.directory = directory;
-        }
-
-        @Override
-        protected StoreStats refresh() {
-            try {
-                return new StoreStats(estimateSize(directory));
-            } catch (IOException ex) {
-                throw new ElasticsearchException("failed to refresh store stats", ex);
-            }
-        }
-
-        private static long estimateSize(Directory directory) throws IOException {
-            long estimatedSize = 0;
-            String[] files = directory.listAll();
-            for (String file : files) {
-                try {
-                    estimatedSize += directory.fileLength(file);
-                } catch (NoSuchFileException | FileNotFoundException | AccessDeniedException e) {
-                    // ignore, the file is not there no more; on Windows, if one thread concurrently deletes a file while
-                    // calling Files.size, you can also sometimes hit AccessDeniedException
-                }
-            }
-            return estimatedSize;
-        }
-    }
-
     /**
      * creates an empty lucene index and a corresponding empty translog. Any existing data will be deleted.
      */

server/src/test/java/org/elasticsearch/index/store/ByteSizeCachingDirectoryTests.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.store;
+
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FilterDirectory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexOutput;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.test.ESTestCase;
+
+import java.io.IOException;
+
+public class ByteSizeCachingDirectoryTests extends ESTestCase {
+
+    private static class LengthCountingDirectory extends FilterDirectory {
+
+        int numFileLengthCalls;
+
+        LengthCountingDirectory(Directory in) {
+            super(in);
+        }
+
+        @Override
+        public long fileLength(String name) throws IOException {
+            numFileLengthCalls++;
+            return super.fileLength(name);
+        }
+    }
+
+    public void testBasics() throws IOException {
+        try (Directory dir = newDirectory()) {
+            try (IndexOutput out = dir.createOutput("quux", IOContext.DEFAULT)) {
+                out.writeBytes(new byte[11], 11);
+            }
+            LengthCountingDirectory countingDir = new LengthCountingDirectory(dir);
+
+            ByteSizeCachingDirectory cachingDir = new ByteSizeCachingDirectory(countingDir, new TimeValue(0));
+            assertEquals(11, cachingDir.estimateSize());
+            assertEquals(11, cachingDir.estimateSize());
+            assertEquals(1, countingDir.numFileLengthCalls);
+
+            try (IndexOutput out = cachingDir.createOutput("foo", IOContext.DEFAULT)) {
+                out.writeBytes(new byte[5], 5);
+            }
+
+            assertEquals(16, cachingDir.estimateSize());
+            assertEquals(3, countingDir.numFileLengthCalls);
+            assertEquals(16, cachingDir.estimateSize());
+            assertEquals(3, countingDir.numFileLengthCalls);
+
+            try (IndexOutput out = cachingDir.createTempOutput("bar", "baz", IOContext.DEFAULT)) {
+                out.writeBytes(new byte[4], 4);
+            }
+
+            assertEquals(20, cachingDir.estimateSize());
+            // +3 because there are 3 files
+            assertEquals(6, countingDir.numFileLengthCalls);
+            assertEquals(20, cachingDir.estimateSize());
+            assertEquals(6, countingDir.numFileLengthCalls);
+        }
+    }
+
+}