[ASTERIXDB-2046][STO] Retry Reading a Page on ClosedChannelException
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Currently when concurrent threads are reading from the same file
and one of them is interrupted, the file channel will be closed and
other threads will fail with AsynchronousCloseException if they were
reading or ClosedChannelException if they were about to read the file.
This change make the threads that were not interrupted retry the read
operation.
Change-Id: I4f9de9f51596314d17e4c6a4a58333e8fd6c03d1
Reviewed-on: https://asterix-gerrit.ics.uci.edu/2005
Sonar-Qube: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: abdullah alamoudi <bamousaa@gmail.com>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
index cf83bca..68e7cd1 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/ErrorCode.java
@@ -127,6 +127,7 @@
public static final int TIMEOUT = 91;
public static final int JOB_HAS_BEEN_CLEARED_FROM_HISTORY = 92;
public static final int JOB_HAS_NOT_BEEN_CREATED_YET = 93;
+ public static final int CANNOT_READ_CLOSED_FILE = 94;
// Compilation error codes.
public static final int RULECOLLECTION_NOT_INSTANCE_OF_LIST = 10000;
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties b/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
index 56abee5..ede2f86 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/resources/errormsg/en.properties
@@ -110,5 +110,6 @@
91 = Operation timed out
92 = Job %1$s has been cleared from job history
93 = Job %1$s has not been created yet
+94 = Cannot read closed file (%1$s)
10000 = The given rule collection %1$s is not an instance of the List class.
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/io/IOManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/io/IOManager.java
index 3da30ab..54a171d 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/io/IOManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/io/IOManager.java
@@ -23,6 +23,7 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;
+import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -209,6 +210,8 @@
// re-open the closed channel. The channel will be closed during the typical file lifecycle
((FileHandle) fHandle).ensureOpen();
throw HyracksDataException.create(e);
+ } catch (ClosedChannelException e) {
+ throw HyracksDataException.create(ErrorCode.CANNOT_READ_CLOSED_FILE, e, fHandle.getFileReference());
} catch (IOException e) {
throw HyracksDataException.create(e);
}
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 f89f638..6304a9a 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
@@ -39,6 +39,7 @@
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IFileHandle;
@@ -58,6 +59,8 @@
private static final int PIN_MAX_WAIT_TIME = 50;
private static final int PIN_ATTEMPT_CYCLES_WARNING_THRESHOLD = 3;
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;
public static final boolean DEBUG = false;
private final int pageSize;
@@ -209,7 +212,7 @@
// disk.
synchronized (cPage) {
if (!cPage.valid) {
- read(cPage);
+ tryRead(cPage);
cPage.valid = true;
}
}
@@ -527,6 +530,33 @@
return false;
}
+ private void tryRead(CachedPage cPage) throws HyracksDataException {
+ for (int i = 1; i <= MAX_PAGE_READ_ATTEMPTS; i++) {
+ try {
+ read(cPage);
+ return;
+ } catch (HyracksDataException readException) {
+ if (readException.getErrorCode() == ErrorCode.CANNOT_READ_CLOSED_FILE && i <= MAX_PAGE_READ_ATTEMPTS) {
+ /**
+ * if the read failure was due to another thread closing the file channel because
+ * it was interrupted, we will try to read again since the interrupted thread
+ * will re-open the file.
+ */
+ try {
+ Thread.sleep(PERIOD_BETWEEN_READ_ATTEMPTS);
+ LOGGER.log(Level.WARNING, String.format("Failed to read page. Retrying attempt (%d/%d)", i + 1,
+ MAX_PAGE_READ_ATTEMPTS), readException);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw HyracksDataException.create(e);
+ }
+ } else {
+ throw readException;
+ }
+ }
+ }
+ }
+
private void read(CachedPage cPage) throws HyracksDataException {
BufferedFileHandle fInfo = getFileInfo(cPage);
cPage.buffer.clear();
@@ -1122,7 +1152,7 @@
ICachedPageInternal old = cachedPages.set(victim.cpid, null);
if (DEBUG) {
assert old == victim;
- } ;
+ }
}
return true;
}
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 e688ee5..e34ee0e 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
@@ -26,12 +26,14 @@
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hyracks.api.context.IHyracksTaskContext;
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.CachedPage;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.storage.common.buffercache.ICachedPage;
import org.apache.hyracks.storage.common.file.BufferedFileHandle;
@@ -314,6 +316,56 @@
bufferCache.close();
}
+ @Test
+ public void interruptedConcurrentReadTest() throws Exception {
+ TestStorageManagerComponentHolder.init(PAGE_SIZE, 200, MAX_OPEN_FILES);
+ IBufferCache bufferCache =
+ TestStorageManagerComponentHolder.getBufferCache(ctx.getJobletContext().getServiceContext());
+ IIOManager ioManager = TestStorageManagerComponentHolder.getIOManager();
+ String fileName = getFileName();
+ FileReference file = ioManager.resolve(fileName);
+ int fileId = bufferCache.createFile(file);
+ int testPageId = 0;
+ bufferCache.openFile(fileId);
+
+ final int expectedPinCount = 100;
+ final AtomicInteger actualPinCount = new AtomicInteger(0);
+ Thread innocentReader = new Thread(() -> {
+ Thread interruptedReader = null;
+ try {
+ for (int i = 0; i < expectedPinCount; i++) {
+ ICachedPage aPage = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId), false);
+ bufferCache.unpin(aPage);
+ ((CachedPage) aPage).invalidate();
+ actualPinCount.incrementAndGet();
+ if (i % 10 == 0) {
+ // start an interruptedReader that will cause the channel to closed
+ interruptedReader = new Thread(() -> {
+ try {
+ Thread.currentThread().interrupt();
+ bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, testPageId + 1), false);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+ interruptedReader.start();
+ }
+ }
+ if (interruptedReader != null) {
+ interruptedReader.join();
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+ innocentReader.start();
+ innocentReader.join();
+ // make sure that all reads by the innocentReader succeeded
+ Assert.assertEquals(actualPinCount.get(), expectedPinCount);
+ // close file
+ bufferCache.closeFile(fileId);
+ }
+
@AfterClass
public static void cleanup() throws Exception {
for (String s : openedFiles) {