You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mb...@apache.org on 2026/10/02 18:54:00 UTC
(asterixdb) branch master updated: [ASTERIXDB-3893][STO] Bound open file descriptors
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new 23facea5bf [ASTERIXDB-3893][STO] Bound open file descriptors
23facea5bf is described below
commit 23facea5bf8cec670de828535b6879712b41ba2f
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Thu Sep 24 10:01:23 2026 -0400
[ASTERIXDB-3893][STO] Bound open file descriptors
- user model changes: no
- storage format changes: no
- interface changes: yes
Details: A buffer cache file holds its OS descriptor for as long as it
is open, and an LSM index keeps every live disk component's files open
for as long as the index is, so descriptors grow with the number of live
components rather than with the files actually in use. With many
datasets that reaches the process's open file limit, and the next file
opened- typically a flush's new component- fails with "Too many open
files" and halts the node. storage.buffercache.maxopenfiles cannot
prevent it: it only closes files nobody references.
Separate the descriptor from the file's lifecycle in the cache. A file
whose descriptor is released stays open to the cache, keeping its id,
pages and reference count, and the next read, write or force reopens it
with the mode it was opened in. Each I/O holds a per-file lock shared
and a release takes it exclusively without waiting, so no descriptor is
released under an I/O in flight. Only a file whose writes have all been
forced is released, which keeps files still being written open and
leaves no write-back error to be lost across a close and reopen.
The cache counts the descriptors its files hold, and once the count
passes storage.buffercache.maxopendescriptors it schedules a pass off
the I/O path that releases the least recently used releasable
descriptors down to nine tenths of the bound. The bound is soft: when
nothing is releasable it is exceeded rather than failing, and the pass
backs off for a second. The default, 0, is half the process's open file
limit, or unbounded where that limit cannot be determined. maxopenfiles
is unchanged.
Each release, reopen and retirement is logged at debug, naming the file,
how long it had been idle or released, and the count left open; each
pass logs the cumulative release and reopen counts.
For publishing as metrics, the cache also counts the descriptors opened
and closed and its release passes, separating those left over the bound;
exposes the bound and the number of files it holds open; and reports
each release, reopen and pass, with its duration, to an optional
IFileDescriptorListener.
On this branch the file handles read and wrote pages on the OS handle
directly; route those through the handle's descriptor lock as well. The
cloud read, persist and sweep paths still use the OS handle outside the
lock, so the bound is not applied when cloud storage is configured.
(cherry picked from commit 3f572744f6baeed4bd3aa831fd6186d83d6d872a)
Ext-ref: MB-74198
Generated-by: Claude Opus 5.5 <no...@anthropic.com>
Change-Id: I0e400c27eca2dce04b6b8f5079d11a6c037f166f
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21976
Reviewed-by: Murtadha Hubail <mh...@apache.org>
Integration-Tests: Michael Blow <mb...@apache.org>
Tested-by: Michael Blow <mb...@apache.org>
---
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 15 +-
.../asterix/common/config/StorageProperties.java | 22 ++
.../asterix-doc/src/site/markdown/ncservice.md | 1 +
.../buffercache/AbstractBufferedFileIOManager.java | 257 ++++++++++++++++++++-
.../storage/common/buffercache/BufferCache.java | 248 +++++++++++++++++++-
.../buffercache/IFileDescriptorListener.java | 61 +++++
.../storage/common/file/BufferedFileHandle.java | 7 +-
.../common/file/CompressedBufferedFileHandle.java | 9 +-
.../hyracks/storage/common/BufferCacheTest.java | 199 ++++++++++++++++
9 files changed, 792 insertions(+), 27 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 434bc1d6bc..bb18fa6dcd 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -300,6 +300,13 @@ public class NCAppRuntimeContext implements INcApplicationContext {
this.ncServiceContext);
receptionist = receptionistFactory.create();
+ // On cloud storage the descriptor bound stays off: the cloud read, persist and sweep paths do I/O on a
+ // file's OS handle directly, outside the buffer cache's descriptor lock, so a release could race them.
+ final int maxOpenDescriptors =
+ cloudConfigurator != null ? Integer.MAX_VALUE : storageProperties.getBufferCacheMaxOpenDescriptors();
+ if (cloudConfigurator != null && storageProperties.getBufferCacheMaxOpenDescriptors() != Integer.MAX_VALUE) {
+ LOGGER.info("not bounding open file descriptors on cloud storage");
+ }
Map<Integer, BufferedFileHandle> fileInfoMap = new HashMap<>();
if (replicationProperties.isReplicationEnabled()) {
if (LOGGER.isInfoEnabled()) {
@@ -318,12 +325,12 @@ public class NCAppRuntimeContext implements INcApplicationContext {
replicationChannel = new ReplicationChannel(this);
bufferCache = new BufferCache(persistenceIOManager, prs, pcp, new FileMapManager(),
- storageProperties.getBufferCacheMaxOpenFiles(), ioQueueLen, getServiceContext().getThreadFactory(),
- replicationManager, fileInfoMap);
+ storageProperties.getBufferCacheMaxOpenFiles(), maxOpenDescriptors, ioQueueLen,
+ getServiceContext().getThreadFactory(), replicationManager, fileInfoMap);
} else {
bufferCache = new BufferCache(persistenceIOManager, prs, pcp, new FileMapManager(),
- storageProperties.getBufferCacheMaxOpenFiles(), ioQueueLen, getServiceContext().getThreadFactory(),
- fileInfoMap, defaultContext);
+ storageProperties.getBufferCacheMaxOpenFiles(), maxOpenDescriptors, ioQueueLen,
+ getServiceContext().getThreadFactory(), fileInfoMap, defaultContext);
}
if (storageProperties.getColumnBufferPoolMaxMemory() <= INVALID_BUFFER_POOL_MAX_MEMORY) {
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java
index 5b34bc2a08..31a390f506 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/StorageProperties.java
@@ -46,6 +46,7 @@ import org.apache.hyracks.api.config.IOption;
import org.apache.hyracks.api.config.IOptionType;
import org.apache.hyracks.api.config.Section;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.MXHelper;
import org.apache.hyracks.util.StorageUtil;
public class StorageProperties extends AbstractProperties {
@@ -55,6 +56,7 @@ public class StorageProperties extends AbstractProperties {
// By default, uses 1/4 of the maximum heap size for read cache, i.e., disk buffer cache.
STORAGE_BUFFERCACHE_SIZE(POSITIVE_LONG_BYTE_UNIT, MAX_HEAP_BYTES / 4),
STORAGE_BUFFERCACHE_MAXOPENFILES(NONNEGATIVE_INTEGER, Integer.MAX_VALUE),
+ STORAGE_BUFFERCACHE_MAXOPENDESCRIPTORS(NONNEGATIVE_INTEGER, 0),
STORAGE_MEMORYCOMPONENT_GLOBALBUDGET(POSITIVE_LONG_BYTE_UNIT, MAX_HEAP_BYTES / 4),
STORAGE_MEMORYCOMPONENT_PAGESIZE(POSITIVE_INTEGER_BYTE_UNIT, StorageUtil.getIntSizeInBytes(128, KILOBYTE)),
STORAGE_MEMORYCOMPONENT_NUMCOMPONENTS(POSITIVE_INTEGER, 2),
@@ -137,6 +139,12 @@ public class StorageProperties extends AbstractProperties {
+ " of the buffer cache page size.";
case STORAGE_BUFFERCACHE_MAXOPENFILES:
return "The maximum number of open files in the buffer cache";
+ case STORAGE_BUFFERCACHE_MAXOPENDESCRIPTORS:
+ return "The number of OS file descriptors the buffer cache aims to hold at most. Beyond it, the"
+ + " descriptors of the least recently used files with no I/O in progress and no unforced"
+ + " writes are closed, and reopened on their next I/O. 0 means half of the process's"
+ + " maximum number of open files, or unbounded where that limit cannot be determined. Not"
+ + " applied on cloud storage";
case STORAGE_MEMORYCOMPONENT_GLOBALBUDGET:
return "The size of memory allocated to the memory components. The value should be a multiple "
+ "of the memory component page size";
@@ -262,6 +270,20 @@ public class StorageProperties extends AbstractProperties {
return accessor.getInt(Option.STORAGE_BUFFERCACHE_MAXOPENFILES);
}
+ /**
+ * @return the configured descriptor bound, or when it is 0, half the process's open file limit, leaving the rest
+ * for sockets, logs and everything else the process opens; {@link Integer#MAX_VALUE} (unbounded) if that
+ * limit cannot be determined
+ */
+ public int getBufferCacheMaxOpenDescriptors() {
+ int configured = accessor.getInt(Option.STORAGE_BUFFERCACHE_MAXOPENDESCRIPTORS);
+ if (configured > 0) {
+ return configured;
+ }
+ long processLimit = MXHelper.getMaxOpenFileCount();
+ return processLimit > 1 ? (int) Math.min(processLimit / 2, Integer.MAX_VALUE) : Integer.MAX_VALUE;
+ }
+
public int getMemoryComponentPageSize() {
return accessor.getInt(Option.STORAGE_MEMORYCOMPONENT_PAGESIZE);
}
diff --git a/asterixdb/asterix-doc/src/site/markdown/ncservice.md b/asterixdb/asterix-doc/src/site/markdown/ncservice.md
index 2352c02295..2f16b18362 100644
--- a/asterixdb/asterix-doc/src/site/markdown/ncservice.md
+++ b/asterixdb/asterix-doc/src/site/markdown/ncservice.md
@@ -324,6 +324,7 @@ The following parameters for slave processes, under "[nc]" sections.
| nc | result.public.port | Public IP port to announce dataset result distribution listener | same as result.listen.port |
| nc | result.sweep.threshold | The duration within which an instance of the result cleanup should be invoked in milliseconds | 60000 |
| nc | result.ttl | Limits the amount of time results for asynchronous jobs should be retained by the system in milliseconds | 86400000 |
+| nc | storage.buffercache.maxopendescriptors | The number of OS file descriptors the buffer cache aims to hold at most. Beyond it, the descriptors of the least recently used files with no I/O in progress and no unforced writes are closed, and reopened on their next I/O. 0 means half of the process's maximum number of open files, or unbounded where that limit cannot be determined. Not applied on cloud storage | 0 |
| nc | storage.buffercache.maxopenfiles | The maximum number of open files in the buffer cache | 2147483647 |
| nc | storage.buffercache.pagesize | The page size in bytes for pages in the buffer cache | 131072 (128 kB) |
| nc | storage.buffercache.size | The size of memory allocated to the disk buffer cache. The value should be a multiple of the buffer cache page size. | 1/4 of the JVM allocated memory |
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/AbstractBufferedFileIOManager.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/AbstractBufferedFileIOManager.java
index c211573012..1421b2145e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/AbstractBufferedFileIOManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/AbstractBufferedFileIOManager.java
@@ -20,12 +20,16 @@ package org.apache.hyracks.storage.common.buffercache;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.StampedLock;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IFileHandle;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.util.IoUtil;
+import org.apache.hyracks.control.nc.io.FileHandle;
import org.apache.hyracks.control.nc.io.IOManager;
import org.apache.hyracks.storage.common.buffercache.context.IBufferCacheReadContext;
import org.apache.hyracks.storage.common.buffercache.context.IBufferCacheWriteContext;
@@ -33,12 +37,20 @@ import org.apache.hyracks.storage.common.compression.file.CompressedFileReferenc
import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter;
import org.apache.hyracks.util.IThreadStats;
import org.apache.hyracks.util.annotations.NotThreadSafe;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
/**
* Handles all IO operations for a specified file.
+ * <p>
+ * The OS descriptor behind the file is separate from the file's lifecycle in the {@link BufferCache}: an open file
+ * whose descriptor has been released by {@link #tryReleaseDescriptor()} stays open as far as the cache is concerned,
+ * keeping its file id, pages and reference count, and has its descriptor reopened by the next read, write or force.
+ * Every such operation holds the descriptor lock shared, so a descriptor is never released under an I/O in flight.
*/
@NotThreadSafe
public abstract class AbstractBufferedFileIOManager {
+ private static final Logger LOGGER = LogManager.getLogger();
private static final String ERROR_MESSAGE = "%s unexpected number of bytes: [expected: %d, actual: %d, file: %s]";
private static final String READ = "Read";
private static final String WRITE = "Written";
@@ -51,6 +63,24 @@ public abstract class AbstractBufferedFileIOManager {
private IFileHandle fileHandle;
private volatile boolean hasOpen;
+ // shared by every I/O on the descriptor, exclusive to release it; not reentrant, and no I/O here nests another
+ // on the same file
+ private final StampedLock descriptorLock = new StampedLock();
+ // serializes reopening among the readers sharing descriptorLock; private, because the buffer cache synchronizes
+ // on the handle itself while closing it, which waits on descriptorLock
+ private final Object descriptorMonitor = new Object();
+ // whether the OS descriptor is currently held; changed under descriptorLock exclusively, or under descriptorLock
+ // shared plus descriptorMonitor when reopening
+ private volatile boolean descriptorOpen;
+ // set once the file is closed, purged or deleted, after which its descriptor is never reopened
+ private volatile boolean retired;
+ // writes issued, and the count covered by the last successful force: a descriptor is only released when equal
+ private final AtomicLong writeCount = new AtomicLong();
+ private volatile long forcedWriteCount;
+ private volatile long lastIoNanos;
+ // System.nanoTime() as of the last release of the descriptor, for logging how long it stayed released
+ private volatile long releasedNanos;
+
protected AbstractBufferedFileIOManager(BufferCache bufferCache, IIOManager ioManager,
BlockingQueue<BufferCacheHeaderHelper> headerPageCache, IPageReplacementStrategy pageReplacementStrategy) {
this.bufferCache = bufferCache;
@@ -112,9 +142,18 @@ public abstract class AbstractBufferedFileIOManager {
* @throws HyracksDataException
*/
public void open(FileReference fileRef) throws HyracksDataException {
- fileHandle = ioManager.open(fileRef, IIOManager.FileReadWriteMode.READ_WRITE,
- IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
- hasOpen = true;
+ final long stamp = descriptorLock.writeLock();
+ try {
+ fileHandle = ioManager.open(fileRef, IIOManager.FileReadWriteMode.READ_WRITE,
+ IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
+ lastIoNanos = System.nanoTime();
+ retired = false;
+ descriptorOpen = true;
+ hasOpen = true;
+ } finally {
+ descriptorLock.unlockWrite(stamp);
+ }
+ bufferCache.descriptorOpened();
}
/**
@@ -124,14 +163,18 @@ public abstract class AbstractBufferedFileIOManager {
*/
public void close() throws HyracksDataException {
if (hasOpen) {
- ioManager.close(fileHandle);
+ retireDescriptor("closed");
}
}
public void purge() throws HyracksDataException {
- ioManager.close(fileHandle);
+ retireDescriptor("purged");
}
+ /**
+ * @return the file's OS handle. I/O done on it directly bypasses the descriptor lock, so a caller doing so must
+ * only run where the buffer cache does not release descriptors
+ */
public IFileHandle getFileHandle() {
return fileHandle;
}
@@ -143,7 +186,146 @@ public abstract class AbstractBufferedFileIOManager {
* @throws HyracksDataException
*/
public void force(boolean metadata) throws HyracksDataException {
- ioManager.sync(fileHandle, metadata);
+ // sampled before the sync: a write that lands during it is not known to be covered by it
+ final long writesCovered = writeCount.get();
+ if (!descriptorOpen && writesCovered == forcedWriteCount) {
+ // released only once everything written was forced, and nothing has been written since
+ return;
+ }
+ final long stamp = beginIo();
+ try {
+ ioManager.sync(fileHandle, metadata);
+ } finally {
+ endIo(stamp);
+ }
+ synchronized (descriptorMonitor) {
+ if (writesCovered > forcedWriteCount) {
+ forcedWriteCount = writesCovered;
+ }
+ }
+ }
+
+ /* ********************************
+ * OS descriptor management
+ * ********************************
+ */
+
+ /**
+ * Release this file's OS descriptor if nothing is using it and everything written through it has been forced.
+ * The file stays open in the {@link BufferCache}; the next I/O reopens the descriptor. Never blocks: a descriptor
+ * with an I/O in flight is left alone.
+ *
+ * @return true if a descriptor was released
+ */
+ public boolean tryReleaseDescriptor() {
+ if (!isDescriptorReleasable()) {
+ return false;
+ }
+ final long stamp = descriptorLock.tryWriteLock();
+ if (stamp == 0L) {
+ return false;
+ }
+ try {
+ if (!isDescriptorReleasable()) {
+ return false;
+ }
+ ioManager.close(fileHandle);
+ descriptorOpen = false;
+ releasedNanos = System.nanoTime();
+ } catch (HyracksDataException e) {
+ // leave the descriptor as it was; it is not in use, and the next pass will try again
+ LOGGER.debug("failed to release the file descriptor of {}", fileHandle.getFileReference(), e);
+ return false;
+ } finally {
+ descriptorLock.unlockWrite(stamp);
+ }
+ final long idleNanos = releasedNanos - lastIoNanos;
+ final int open = bufferCache.descriptorReleased(idleNanos);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("released the file descriptor of {} after {}ms idle; {} now open", getFileReference(),
+ TimeUnit.NANOSECONDS.toMillis(idleNanos), open);
+ }
+ return true;
+ }
+
+ /**
+ * @return whether the OS descriptor is currently held
+ */
+ public final boolean isDescriptorOpen() {
+ return descriptorOpen;
+ }
+
+ /**
+ * @return {@link System#nanoTime()} as of the last I/O through this file, or of its opening
+ */
+ public final long getLastIoNanos() {
+ return lastIoNanos;
+ }
+
+ private boolean isDescriptorReleasable() {
+ return descriptorOpen && !retired && fileHandle != null && writeCount.get() == forcedWriteCount;
+ }
+
+ private void retireDescriptor(String reason) throws HyracksDataException {
+ // exclusive, and blocking: the file is being closed for good, so wait out any I/O still in flight
+ final long stamp = descriptorLock.writeLock();
+ final IFileHandle retiredHandle = fileHandle;
+ boolean wasOpen;
+ try {
+ retired = true;
+ wasOpen = descriptorOpen;
+ descriptorOpen = false;
+ if (fileHandle != null) {
+ ioManager.close(fileHandle);
+ }
+ } finally {
+ descriptorLock.unlockWrite(stamp);
+ }
+ final int open = wasOpen ? bufferCache.descriptorClosed() : bufferCache.getOpenDescriptorCount();
+ if (retiredHandle != null && LOGGER.isDebugEnabled()) {
+ LOGGER.debug("retired the file descriptor of {} ({}; descriptor {}); {} now open",
+ retiredHandle.getFileReference(), reason, wasOpen ? "held" : "already released", open);
+ }
+ }
+
+ private long beginIo() throws HyracksDataException {
+ final long stamp = descriptorLock.readLock();
+ try {
+ if (!descriptorOpen) {
+ reopenDescriptor();
+ }
+ lastIoNanos = System.nanoTime();
+ return stamp;
+ } catch (Throwable th) {
+ descriptorLock.unlockRead(stamp);
+ throw th;
+ }
+ }
+
+ private void endIo(long stamp) {
+ descriptorLock.unlockRead(stamp);
+ }
+
+ private void reopenDescriptor() throws HyracksDataException {
+ boolean reopened = false;
+ long reopenedNanos = 0L;
+ // several readers may get here at once under the shared lock; only one of them reopens
+ synchronized (descriptorMonitor) {
+ if (!descriptorOpen && !retired && fileHandle != null) {
+ ((FileHandle) fileHandle).ensureOpen();
+ descriptorOpen = true;
+ reopened = true;
+ reopenedNanos = System.nanoTime();
+ }
+ }
+ if (reopened) {
+ final long releasedForNanos = reopenedNanos - releasedNanos;
+ final int open = bufferCache.descriptorReopened(releasedForNanos);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("reopened the file descriptor of {} after {}ms released; {} now open", getFileReference(),
+ TimeUnit.NANOSECONDS.toMillis(releasedForNanos), open);
+ }
+ }
}
/**
@@ -162,6 +344,7 @@ public abstract class AbstractBufferedFileIOManager {
public abstract int getNumberOfPages() throws HyracksDataException;
public void markAsDeleted() throws HyracksDataException {
+ retired = true;
fileHandle = null;
}
@@ -286,15 +469,71 @@ public abstract class AbstractBufferedFileIOManager {
}
protected final long readToBuffer(ByteBuffer buf, long offset) throws HyracksDataException {
- return ioManager.syncRead(fileHandle, offset, buf);
+ final long stamp = beginIo();
+ try {
+ return ioManager.syncRead(fileHandle, offset, buf);
+ } finally {
+ endIo(stamp);
+ }
+ }
+
+ /**
+ * Read a page's header from the file, under the descriptor lock.
+ */
+ protected final long readHeaderFromFile(BufferCacheHeaderHelper header, long offset, int size)
+ throws HyracksDataException {
+ final long stamp = beginIo();
+ try {
+ return header.readFromFile(ioManager, fileHandle, offset, size);
+ } finally {
+ endIo(stamp);
+ }
+ }
+
+ /**
+ * Write through the given context, under the descriptor lock.
+ */
+ protected final long writeToFile(IBufferCacheWriteContext context, long offset, ByteBuffer buf)
+ throws HyracksDataException {
+ // counted before the write, so a force that overlaps it cannot claim to cover it
+ writeCount.incrementAndGet();
+ final long stamp = beginIo();
+ try {
+ return context.write(ioManager, fileHandle, offset, buf);
+ } finally {
+ endIo(stamp);
+ }
+ }
+
+ protected final long writeToFile(IBufferCacheWriteContext context, long offset, ByteBuffer[] buf)
+ throws HyracksDataException {
+ writeCount.incrementAndGet();
+ final long stamp = beginIo();
+ try {
+ return context.write(ioManager, fileHandle, offset, buf);
+ } finally {
+ endIo(stamp);
+ }
}
protected final long writeExtraToFile(ByteBuffer buf, long offset) throws HyracksDataException {
- return ioManager.doSyncWrite(fileHandle, offset, buf);
+ writeCount.incrementAndGet();
+ final long stamp = beginIo();
+ try {
+ return ioManager.doSyncWrite(fileHandle, offset, buf);
+ } finally {
+ endIo(stamp);
+ }
}
protected final long writeExtraToFile(ByteBuffer[] buf, long offset) throws HyracksDataException {
- return ioManager.doSyncWrite(fileHandle, offset, buf);
+ writeCount.incrementAndGet();
+ final long stamp = beginIo();
+ try {
+ return ioManager.doSyncWrite(fileHandle, offset, buf);
+ } finally {
+ endIo(stamp);
+ }
}
protected final long getFileSize() throws HyracksDataException {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
index 5f022098f7..ef99408f8b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -31,7 +32,11 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -69,11 +74,16 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
private static final int MAX_PIN_ATTEMPT_CYCLES = 1000;
private static final int MAX_PAGE_READ_ATTEMPTS = 5;
private static final long PERIOD_BETWEEN_READ_ATTEMPTS = 100;
+ // a pass that cannot get under the descriptor bound is not repeated for this long
+ private static final long DESCRIPTOR_RELEASE_BACKOFF_NANOS = TimeUnit.SECONDS.toNanos(1);
+ private static final long DESCRIPTOR_WARN_INTERVAL_NANOS = TimeUnit.MINUTES.toNanos(1);
public static final boolean DEBUG = false;
private final int pageSize;
private final int maxOpenFiles;
private final IBufferCacheReadContext defaultContext;
+ private final int maxOpenDescriptors;
+ private final int descriptorLowWatermark;
final IIOManager ioManager;
private final CacheBucket[] pageMap;
private final IPageReplacementStrategy pageReplacementStrategy;
@@ -87,6 +97,18 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
private final List<ICachedPageInternal> cachedPages = new ArrayList<>();
private final AtomicLong masterPinCount = new AtomicLong();
private final Map<Thread, IThreadStats> statsSubscribers = new ConcurrentHashMap<>();
+ private final Executor executor;
+ private final AtomicInteger openDescriptors = new AtomicInteger();
+ private final AtomicLong descriptorsOpened = new AtomicLong();
+ private final AtomicLong descriptorsReleased = new AtomicLong();
+ private final AtomicLong descriptorsReopened = new AtomicLong();
+ private final AtomicLong descriptorsClosed = new AtomicLong();
+ private final AtomicLong descriptorReleasePasses = new AtomicLong();
+ private final AtomicLong descriptorReleasePassesOverBound = new AtomicLong();
+ private volatile IFileDescriptorListener descriptorListener = IFileDescriptorListener.NO_OP;
+ private final AtomicBoolean descriptorReleaseScheduled = new AtomicBoolean();
+ private volatile long nextDescriptorReleaseNanos = System.nanoTime();
+ private volatile long nextDescriptorWarnNanos = System.nanoTime();
private boolean closed;
@@ -101,11 +123,35 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
IPageCleanerPolicy pageCleanerPolicy, IFileMapManager fileMapManager, int maxOpenFiles, int ioQueuelen,
ThreadFactory threadFactory, Map<Integer, BufferedFileHandle> fileInfoMap,
IBufferCacheReadContext defaultContext) {
+ this(ioManager, pageReplacementStrategy, pageCleanerPolicy, fileMapManager, maxOpenFiles, Integer.MAX_VALUE,
+ ioQueuelen, threadFactory, fileInfoMap, defaultContext);
+ }
+
+ /**
+ * @param maxOpenFiles
+ * the maximum number of files the cache holds open; opening a file beyond it closes an unreferenced
+ * one, and fails if every open file is referenced
+ * @param maxOpenDescriptors
+ * the number of OS file descriptors the cache aims to hold at most; beyond it, the descriptors of the
+ * least recently used files which have no I/O in flight and nothing written since their last force are
+ * released, and reopened on their next I/O. The bound is soft: when nothing is releasable it is
+ * exceeded rather than failing. {@link Integer#MAX_VALUE} disables releasing.
+ */
+ public BufferCache(IIOManager ioManager, IPageReplacementStrategy pageReplacementStrategy,
+ IPageCleanerPolicy pageCleanerPolicy, IFileMapManager fileMapManager, int maxOpenFiles,
+ int maxOpenDescriptors, int ioQueuelen, ThreadFactory threadFactory,
+ Map<Integer, BufferedFileHandle> fileInfoMap, IBufferCacheReadContext defaultContext) {
+ if (maxOpenDescriptors <= 0) {
+ throw new IllegalArgumentException("maxOpenDescriptors must be positive: " + maxOpenDescriptors);
+ }
this.headerPageCache = new ArrayBlockingQueue<>(ioQueuelen);
this.ioManager = ioManager;
this.pageSize = pageReplacementStrategy.getPageSize();
this.maxOpenFiles = maxOpenFiles;
this.defaultContext = defaultContext;
+ this.maxOpenDescriptors = maxOpenDescriptors;
+ // release a tenth of the bound at a time, so passes are rare
+ this.descriptorLowWatermark = maxOpenDescriptors - Math.max(1, maxOpenDescriptors / 10);
pageReplacementStrategy.setBufferCache(this);
pageMap = new CacheBucket[pageReplacementStrategy.getMaxAllowedNumPages() * MAP_FACTOR + 1];
for (int i = 0; i < pageMap.length; ++i) {
@@ -115,7 +161,7 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
this.pageCleanerPolicy = pageCleanerPolicy;
this.fileMapManager = fileMapManager;
- Executor executor = Executors.newCachedThreadPool(threadFactory);
+ executor = Executors.newCachedThreadPool(threadFactory);
this.fileInfoMap = fileInfoMap;
cleanerThread = new CleanerThread();
executor.execute(cleanerThread);
@@ -134,8 +180,17 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
IPageCleanerPolicy pageCleanerPolicy, IFileMapManager fileMapManager, int maxOpenFiles, int ioQueueLen,
ThreadFactory threadFactory, IIOReplicationManager ioReplicationManager,
Map<Integer, BufferedFileHandle> fileInfoMap) {
- this(ioManager, pageReplacementStrategy, pageCleanerPolicy, fileMapManager, maxOpenFiles, ioQueueLen,
- threadFactory, fileInfoMap, DefaultBufferCacheReadContextProvider.DEFAULT);
+ this(ioManager, pageReplacementStrategy, pageCleanerPolicy, fileMapManager, maxOpenFiles, Integer.MAX_VALUE,
+ ioQueueLen, threadFactory, ioReplicationManager, fileInfoMap);
+ }
+
+ //this constructor is used when replication is enabled to pass the IIOReplicationManager
+ public BufferCache(IIOManager ioManager, IPageReplacementStrategy pageReplacementStrategy,
+ IPageCleanerPolicy pageCleanerPolicy, IFileMapManager fileMapManager, int maxOpenFiles,
+ int maxOpenDescriptors, int ioQueueLen, ThreadFactory threadFactory,
+ IIOReplicationManager ioReplicationManager, Map<Integer, BufferedFileHandle> fileInfoMap) {
+ this(ioManager, pageReplacementStrategy, pageCleanerPolicy, fileMapManager, maxOpenFiles, maxOpenDescriptors,
+ ioQueueLen, threadFactory, fileInfoMap, DefaultBufferCacheReadContextProvider.DEFAULT);
this.ioReplicationManager = ioReplicationManager;
}
@@ -455,6 +510,9 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
buffer.append("Number of physical pages: ").append(pageReplacementStrategy.getMaxAllowedNumPages())
.append('\n');
buffer.append("Hash table size: ").append(pageMap.length).append('\n');
+ buffer.append("Open file descriptors: ").append(openDescriptors.get()).append(" (bound ")
+ .append(maxOpenDescriptors).append(", released ").append(descriptorsReleased.get())
+ .append(", reopened ").append(descriptorsReopened.get()).append(")\n");
buffer.append("Page Map:\n");
buffer.append("cpid -> [fileId:pageId, pinCount, valid/invalid, confiscated/physical, dirty/clean]");
int nCachedPages = 0;
@@ -1494,4 +1552,188 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent, I
return fInfo.getCompressedPageWriter();
}
+ /**
+ * @return the number of OS file descriptors currently held by this cache's files
+ */
+ public int getOpenDescriptorCount() {
+ return openDescriptors.get();
+ }
+
+ /**
+ * @return the bound on open descriptors, or {@link Integer#MAX_VALUE} if unbounded
+ */
+ public int getMaxOpenDescriptors() {
+ return maxOpenDescriptors;
+ }
+
+ /**
+ * @return the number of files this cache holds open, whether or not their descriptors are released
+ */
+ public int getOpenFileCount() {
+ synchronized (fileInfoMap) {
+ return fileInfoMap.size();
+ }
+ }
+
+ /**
+ * @return how many descriptors have been opened by opening a file, not counting reopens
+ */
+ public long getOpenedDescriptorCount() {
+ return descriptorsOpened.get();
+ }
+
+ /**
+ * @return how many held descriptors have been closed by closing, purging or deleting their file
+ */
+ public long getClosedDescriptorCount() {
+ return descriptorsClosed.get();
+ }
+
+ /**
+ * @return how many release passes have run
+ */
+ public long getDescriptorReleasePassCount() {
+ return descriptorReleasePasses.get();
+ }
+
+ /**
+ * @return how many release passes have left the open descriptors over the bound, having found too few releasable
+ */
+ public long getDescriptorReleasePassOverBoundCount() {
+ return descriptorReleasePassesOverBound.get();
+ }
+
+ /**
+ * Set the listener to notify of descriptor releases, reopens and release passes; null clears it.
+ */
+ public void setDescriptorListener(IFileDescriptorListener listener) {
+ descriptorListener = listener != null ? listener : IFileDescriptorListener.NO_OP;
+ }
+
+ /**
+ * @return how many times a file's descriptor has been released to keep under the descriptor bound
+ */
+ public long getReleasedDescriptorCount() {
+ return descriptorsReleased.get();
+ }
+
+ /**
+ * @return how many times a released descriptor has been reopened by a subsequent I/O
+ */
+ public long getReopenedDescriptorCount() {
+ return descriptorsReopened.get();
+ }
+
+ // each returns the number of descriptors open afterwards
+
+ int descriptorOpened() {
+ descriptorsOpened.incrementAndGet();
+ return descriptorAcquired();
+ }
+
+ int descriptorReopened(long releasedNanos) {
+ descriptorsReopened.incrementAndGet();
+ final int open = descriptorAcquired();
+ try {
+ descriptorListener.descriptorReopened(releasedNanos);
+ } catch (Exception e) {
+ LOGGER.debug("descriptor listener failed", e);
+ }
+ return open;
+ }
+
+ int descriptorReleased(long idleNanos) {
+ descriptorsReleased.incrementAndGet();
+ final int open = openDescriptors.decrementAndGet();
+ try {
+ descriptorListener.descriptorReleased(idleNanos);
+ } catch (Exception e) {
+ LOGGER.debug("descriptor listener failed", e);
+ }
+ return open;
+ }
+
+ int descriptorClosed() {
+ descriptorsClosed.incrementAndGet();
+ return openDescriptors.decrementAndGet();
+ }
+
+ private int descriptorAcquired() {
+ // called from the open and I/O paths, so it only ever schedules the release pass, never runs it
+ final int open = openDescriptors.incrementAndGet();
+ if (open > maxOpenDescriptors && !closed && System.nanoTime() - nextDescriptorReleaseNanos >= 0
+ && descriptorReleaseScheduled.compareAndSet(false, true)) {
+ try {
+ executor.execute(this::releaseColdDescriptors);
+ } catch (RejectedExecutionException e) {
+ descriptorReleaseScheduled.set(false);
+ }
+ }
+ return open;
+ }
+
+ private record DescriptorCandidate(BufferedFileHandle handle, long idleNanos) {
+ }
+
+ /**
+ * Release the descriptors of the least recently used files until the count is back under the low watermark, or
+ * nothing more is releasable- a file with an I/O in flight, or with writes not yet forced, keeps its descriptor.
+ */
+ private void releaseColdDescriptors() {
+ try {
+ if (closed || openDescriptors.get() <= descriptorLowWatermark) {
+ return;
+ }
+ // snapshot each file's idle time against a single clock reading: the last I/O times move under us as
+ // I/O continues, and nanoTime() values are only comparable as differences, never directly
+ final long snapshotNanos = System.nanoTime();
+ final List<DescriptorCandidate> candidates = new ArrayList<>();
+ synchronized (fileInfoMap) {
+ for (BufferedFileHandle fh : fileInfoMap.values()) {
+ if (fh.isDescriptorOpen()) {
+ candidates.add(new DescriptorCandidate(fh, snapshotNanos - fh.getLastIoNanos()));
+ }
+ }
+ }
+ final int n = candidates.size();
+ candidates.sort(Comparator.comparingLong(DescriptorCandidate::idleNanos).reversed());
+ int released = 0;
+ for (int i = 0; i < n && openDescriptors.get() > descriptorLowWatermark && !closed; i++) {
+ if (candidates.get(i).handle().tryReleaseDescriptor()) {
+ released++;
+ }
+ }
+ final int open = openDescriptors.get();
+ descriptorReleasePasses.incrementAndGet();
+ if (open > maxOpenDescriptors) {
+ descriptorReleasePassesOverBound.incrementAndGet();
+ }
+ try {
+ descriptorListener.releasePassCompleted(released, System.nanoTime() - snapshotNanos,
+ open <= maxOpenDescriptors);
+ } catch (Exception e) {
+ LOGGER.debug("descriptor listener failed", e);
+ }
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(
+ "released {} of {} file descriptors; {} now open against a bound of {} ({} released and {}"
+ + " reopened in total)",
+ released, n, open, maxOpenDescriptors, descriptorsReleased.get(), descriptorsReopened.get());
+ }
+ if (open > maxOpenDescriptors) {
+ final long now = System.nanoTime();
+ nextDescriptorReleaseNanos = now + DESCRIPTOR_RELEASE_BACKOFF_NANOS;
+ if (now - nextDescriptorWarnNanos >= 0) {
+ nextDescriptorWarnNanos = now + DESCRIPTOR_WARN_INTERVAL_NANOS;
+ LOGGER.warn("{} file descriptors open against a bound of {}; the rest are in use or have writes"
+ + " not yet forced", open, maxOpenDescriptors);
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.warn("failure releasing file descriptors", e);
+ } finally {
+ descriptorReleaseScheduled.set(false);
+ }
+ }
+
}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFileDescriptorListener.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFileDescriptorListener.java
new file mode 100644
index 0000000000..5e47bdcba7
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IFileDescriptorListener.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.hyracks.storage.common.buffercache;
+
+/**
+ * Observes the {@link BufferCache}'s management of OS file descriptors, for instance to publish metrics. Called on the
+ * I/O path and the release pass, so implementations must be cheap and must not block; anything thrown is logged and
+ * otherwise ignored.
+ */
+public interface IFileDescriptorListener {
+
+ IFileDescriptorListener NO_OP = new IFileDescriptorListener() {
+ };
+
+ /**
+ * A file's descriptor was released to keep under the descriptor bound.
+ *
+ * @param idleNanos
+ * how long the file had gone without I/O when released
+ */
+ default void descriptorReleased(long idleNanos) {
+ }
+
+ /**
+ * A released descriptor was reopened by an I/O on its file.
+ *
+ * @param releasedNanos
+ * how long the descriptor had been released
+ */
+ default void descriptorReopened(long releasedNanos) {
+ }
+
+ /**
+ * A release pass finished.
+ *
+ * @param released
+ * the descriptors it released
+ * @param durationNanos
+ * how long it took
+ * @param underBound
+ * whether it left the open descriptors within the bound
+ */
+ default void releasePassCompleted(int released, long durationNanos, boolean underBound) {
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/BufferedFileHandle.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/BufferedFileHandle.java
index 3ba1029b18..37452c0594 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/BufferedFileHandle.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/BufferedFileHandle.java
@@ -26,7 +26,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.api.io.IFileHandle;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.storage.common.buffercache.AbstractBufferedFileIOManager;
import org.apache.hyracks.storage.common.buffercache.BufferCache;
@@ -80,9 +79,8 @@ public class BufferedFileHandle extends AbstractBufferedFileIOManager {
final BufferCacheHeaderHelper header = checkoutHeaderHelper();
try {
setPageInfo(cPage);
- IFileHandle handle = getFileHandle();
int pageSize = bufferCache.getPageSizeWithHeader();
- long bytesRead = header.readFromFile(ioManager, handle, getFirstPageOffset(cPage), pageSize);
+ long bytesRead = readHeaderFromFile(header, getFirstPageOffset(cPage), pageSize);
if (!verifyBytesRead(bufferCache.getPageSizeWithHeader(), bytesRead)) {
return;
@@ -117,7 +115,6 @@ public class BufferedFileHandle extends AbstractBufferedFileIOManager {
IBufferCacheWriteContext context) throws HyracksDataException {
final ByteBuffer buf = cPage.getBuffer();
final boolean contiguousLargePages = getPageId(cPage.getDiskPageId()) + 1 == extraBlockPageId;
- IFileHandle handle = getFileHandle();
long bytesWritten;
long offset;
try {
@@ -125,7 +122,7 @@ public class BufferedFileHandle extends AbstractBufferedFileIOManager {
buf.position(0);
ByteBuffer[] buffers = header.prepareWrite(cPage);
offset = getFirstPageOffset(cPage);
- bytesWritten = context.write(ioManager, handle, offset, buffers);
+ bytesWritten = writeToFile(context, offset, buffers);
} finally {
returnHeaderHelper(header);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/CompressedBufferedFileHandle.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/CompressedBufferedFileHandle.java
index 86511d1067..6c7d132fc2 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/CompressedBufferedFileHandle.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/file/CompressedBufferedFileHandle.java
@@ -24,7 +24,6 @@ import java.util.concurrent.BlockingQueue;
import org.apache.hyracks.api.compression.ICompressorDecompressor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.api.io.IFileHandle;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.storage.common.buffercache.BufferCache;
import org.apache.hyracks.storage.common.buffercache.BufferCacheHeaderHelper;
@@ -54,9 +53,8 @@ public class CompressedBufferedFileHandle extends BufferedFileHandle {
final BufferCacheHeaderHelper header = checkoutHeaderHelper();
try {
compressedFileManager.setCompressedPageInfo(cPage);
- IFileHandle handle = getFileHandle();
int size = cPage.getCompressedPageSize();
- long bytesRead = header.readFromFile(ioManager, handle, getFirstPageOffset(cPage), size);
+ long bytesRead = readHeaderFromFile(header, getFirstPageOffset(cPage), size);
if (!verifyBytesRead(cPage.getCompressedPageSize(), bytesRead)) {
return;
}
@@ -103,7 +101,6 @@ public class CompressedBufferedFileHandle extends BufferedFileHandle {
protected void write(CachedPage cPage, BufferCacheHeaderHelper header, int totalPages, int extraBlockPageId,
IBufferCacheWriteContext context) throws HyracksDataException {
try {
- IFileHandle handle = getFileHandle();
final ByteBuffer cBuffer = header.prepareWrite(cPage, getRequiredBufferSize());
final ByteBuffer uBuffer = cPage.getBuffer();
final long pageId = cPage.getDiskPageId();
@@ -117,13 +114,13 @@ public class CompressedBufferedFileHandle extends BufferedFileHandle {
cBuffer.position(0);
offset = compressedFileManager.writePageInfo(pageId, cBuffer.remaining());
expectedBytesWritten = cBuffer.limit();
- bytesWritten = context.write(ioManager, handle, offset, cBuffer);
+ bytesWritten = writeToFile(context, offset, cBuffer);
} else {
// Compression did not gain any savings
final ByteBuffer[] buffers = header.prepareWrite(cPage);
offset = compressedFileManager.writePageInfo(pageId, bufferCache.getPageSizeWithHeader());
expectedBytesWritten = buffers[0].limit() + (long) buffers[1].limit();
- bytesWritten = context.write(ioManager, handle, offset, buffers);
+ bytesWritten = writeToFile(context, offset, buffers);
}
verifyBytesWritten(expectedBytesWritten, bytesWritten);
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java
index c8a95323df..63f096d305 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/BufferCacheTest.java
@@ -41,13 +41,21 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.HyracksException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IIOManager;
+import org.apache.hyracks.storage.common.buffercache.BufferCache;
import org.apache.hyracks.storage.common.buffercache.CachedPage;
+import org.apache.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
+import org.apache.hyracks.storage.common.buffercache.DefaultDiskCachedPageAllocator;
+import org.apache.hyracks.storage.common.buffercache.DelayPageCleanerPolicy;
import org.apache.hyracks.storage.common.buffercache.HaltOnFailureCallback;
+import org.apache.hyracks.storage.common.buffercache.HeapBufferAllocator;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.storage.common.buffercache.IFileDescriptorListener;
import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback;
+import org.apache.hyracks.storage.common.buffercache.context.read.DefaultBufferCacheReadContextProvider;
import org.apache.hyracks.storage.common.buffercache.context.write.DefaultBufferCacheWriteContext;
import org.apache.hyracks.storage.common.file.BufferedFileHandle;
+import org.apache.hyracks.storage.common.file.FileMapManager;
import org.apache.hyracks.test.support.TestStorageManagerComponentHolder;
import org.apache.hyracks.test.support.TestUtils;
import org.apache.logging.log4j.Level;
@@ -469,6 +477,197 @@ public class BufferCacheTest {
Assert.assertEquals(0, aPage.getBuffer().position());
}
+ private BufferCache createDescriptorBoundedCache(int numPages, int maxOpenDescriptors) throws HyracksDataException {
+ TestStorageManagerComponentHolder.init(PAGE_SIZE, numPages, MAX_OPEN_FILES);
+ return new BufferCache(TestStorageManagerComponentHolder.getIOManager(),
+ new ClockPageReplacementStrategy(new HeapBufferAllocator(), DefaultDiskCachedPageAllocator.INSTANCE,
+ PAGE_SIZE, numPages),
+ new DelayPageCleanerPolicy(1000), new FileMapManager(), Integer.MAX_VALUE, maxOpenDescriptors, 10,
+ Thread::new, new HashMap<>(), DefaultBufferCacheReadContextProvider.DEFAULT);
+ }
+
+ /**
+ * Creates and opens a file, writes a page identifying it, and writes that page to disk; forces it if asked.
+ */
+ private int createFileWithPage(BufferCache bufferCache, boolean force) throws HyracksDataException {
+ FileReference file = TestStorageManagerComponentHolder.getIOManager().resolve(getFileName());
+ int fileId = bufferCache.createFile(file);
+ bufferCache.openFile(fileId);
+ ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, 0), NEW);
+ page.acquireWriteLatch();
+ try {
+ for (int i = 0; i < 8; i++) {
+ page.getBuffer().putInt(i * 4, fileId * 31 + i);
+ }
+ } finally {
+ page.releaseWriteLatch(true);
+ }
+ try {
+ bufferCache.flush(page);
+ } finally {
+ bufferCache.unpin(page);
+ }
+ if (force) {
+ bufferCache.force(fileId, true);
+ }
+ return fileId;
+ }
+
+ private static void verifyPage(IBufferCache bufferCache, int fileId) throws HyracksDataException {
+ ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, 0));
+ page.acquireReadLatch();
+ try {
+ for (int i = 0; i < 8; i++) {
+ Assert.assertEquals("file " + fileId + " int " + i, fileId * 31 + i, page.getBuffer().getInt(i * 4));
+ }
+ } finally {
+ page.releaseReadLatch();
+ bufferCache.unpin(page);
+ }
+ }
+
+ private static void awaitTrue(String what, Callable<Boolean> condition) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ while (!condition.call()) {
+ if (System.nanoTime() - deadline > 0) {
+ Assert.fail("timed out waiting for " + what);
+ }
+ TimeUnit.MILLISECONDS.sleep(10);
+ }
+ }
+
+ @Test
+ public void releasedDescriptorsAreReopenedOnNextIoTest() throws Exception {
+ final int maxOpenDescriptors = 4;
+ final int numFiles = 16;
+ // fewer cache pages than files, so that reading the files back must go to disk
+ BufferCache bufferCache = createDescriptorBoundedCache(NUM_PAGES, maxOpenDescriptors);
+ final AtomicInteger listenedReleases = new AtomicInteger();
+ final AtomicInteger listenedReopens = new AtomicInteger();
+ final AtomicInteger listenedPasses = new AtomicInteger();
+ bufferCache.setDescriptorListener(new IFileDescriptorListener() {
+ @Override
+ public void descriptorReleased(long idleNanos) {
+ Assert.assertTrue(idleNanos >= 0);
+ listenedReleases.incrementAndGet();
+ }
+
+ @Override
+ public void descriptorReopened(long releasedNanos) {
+ Assert.assertTrue(releasedNanos >= 0);
+ listenedReopens.incrementAndGet();
+ }
+
+ @Override
+ public void releasePassCompleted(int released, long durationNanos, boolean underBound) {
+ listenedPasses.incrementAndGet();
+ }
+ });
+ Assert.assertEquals(maxOpenDescriptors, bufferCache.getMaxOpenDescriptors());
+ List<Integer> fileIds = new ArrayList<>();
+ for (int i = 0; i < numFiles; i++) {
+ fileIds.add(createFileWithPage(bufferCache, true));
+ }
+ Assert.assertEquals(numFiles, bufferCache.getOpenedDescriptorCount());
+ Assert.assertEquals(numFiles, bufferCache.getOpenFileCount());
+ awaitTrue("descriptors to be released", () -> bufferCache.getOpenDescriptorCount() <= maxOpenDescriptors);
+ Assert.assertTrue(bufferCache.getReleasedDescriptorCount() >= numFiles - maxOpenDescriptors);
+
+ // every file is still open as far as the cache is concerned, and reads correctly
+ for (int fileId : fileIds) {
+ verifyPage(bufferCache, fileId);
+ }
+ Assert.assertTrue("expected reads to reopen released descriptors",
+ bufferCache.getReopenedDescriptorCount() > 0);
+
+ for (int fileId : fileIds) {
+ bufferCache.closeFile(fileId);
+ }
+ bufferCache.close();
+ Assert.assertEquals(0, bufferCache.getOpenDescriptorCount());
+ // every descriptor acquired was given up, either released or closed
+ Assert.assertEquals(bufferCache.getOpenedDescriptorCount() + bufferCache.getReopenedDescriptorCount(),
+ bufferCache.getReleasedDescriptorCount() + bufferCache.getClosedDescriptorCount());
+ Assert.assertEquals(bufferCache.getReleasedDescriptorCount(), listenedReleases.get());
+ Assert.assertEquals(bufferCache.getReopenedDescriptorCount(), listenedReopens.get());
+ // a pass counts itself just before notifying, so one finishing as the cache closes may still be on its way
+ awaitTrue("the listener to see every pass",
+ () -> bufferCache.getDescriptorReleasePassCount() == listenedPasses.get());
+ Assert.assertTrue(listenedPasses.get() > 0);
+ }
+
+ @Test
+ public void unforcedWritesKeepTheirDescriptorTest() throws Exception {
+ final int maxOpenDescriptors = 2;
+ BufferCache bufferCache = createDescriptorBoundedCache(NUM_PAGES, maxOpenDescriptors);
+ List<Integer> fileIds = new ArrayList<>();
+ // the third open goes over the bound, but a file with writes not yet forced must keep its descriptor
+ for (int i = 0; i < 3; i++) {
+ fileIds.add(createFileWithPage(bufferCache, false));
+ }
+ TimeUnit.MILLISECONDS.sleep(200);
+ Assert.assertEquals(0, bufferCache.getReleasedDescriptorCount());
+ Assert.assertEquals(3, bufferCache.getOpenDescriptorCount());
+ Assert.assertEquals(1, bufferCache.getDescriptorReleasePassCount());
+ Assert.assertEquals(1, bufferCache.getDescriptorReleasePassOverBoundCount());
+
+ // once forced they are releasable; a pass that found nothing to release backs off, so wait that out, and
+ // the next open schedules another
+ for (int fileId : fileIds) {
+ bufferCache.force(fileId, true);
+ }
+ TimeUnit.MILLISECONDS.sleep(1100);
+ fileIds.add(createFileWithPage(bufferCache, true));
+ awaitTrue("descriptors to get back under the bound",
+ () -> bufferCache.getOpenDescriptorCount() <= maxOpenDescriptors);
+ Assert.assertTrue(bufferCache.getReleasedDescriptorCount() >= 2);
+
+ for (int fileId : fileIds) {
+ verifyPage(bufferCache, fileId);
+ bufferCache.closeFile(fileId);
+ }
+ bufferCache.close();
+ Assert.assertEquals(0, bufferCache.getOpenDescriptorCount());
+ }
+
+ @Test
+ public void concurrentReadsWhileDescriptorsAreReleasedTest() throws Exception {
+ final int maxOpenDescriptors = 4;
+ final int numFiles = 32;
+ final int numThreads = 8;
+ final int readsPerThread = 2000;
+ BufferCache bufferCache = createDescriptorBoundedCache(NUM_PAGES, maxOpenDescriptors);
+ List<Integer> fileIds = new ArrayList<>();
+ for (int i = 0; i < numFiles; i++) {
+ fileIds.add(createFileWithPage(bufferCache, true));
+ }
+ ExecutorService readers = Executors.newFixedThreadPool(numThreads);
+ try {
+ List<Future<Void>> results = new ArrayList<>();
+ for (int t = 0; t < numThreads; t++) {
+ final Random random = new Random(t);
+ results.add(readers.submit(() -> {
+ for (int i = 0; i < readsPerThread; i++) {
+ verifyPage(bufferCache, fileIds.get(random.nextInt(numFiles)));
+ }
+ return null;
+ }));
+ }
+ for (Future<Void> result : results) {
+ result.get(2, TimeUnit.MINUTES);
+ }
+ } finally {
+ readers.shutdownNow();
+ }
+ Assert.assertTrue(bufferCache.getReleasedDescriptorCount() > 0);
+ Assert.assertTrue(bufferCache.getReopenedDescriptorCount() > 0);
+ for (int fileId : fileIds) {
+ bufferCache.closeFile(fileId);
+ }
+ bufferCache.close();
+ Assert.assertEquals(0, bufferCache.getOpenDescriptorCount());
+ }
+
@AfterClass
public static void cleanup() throws Exception {
for (String s : openedFiles) {