Ian Maxon has uploaded a new change for review.
https://asterix-gerrit.ics.uci.edu/3133
Change subject: [WIP][NO ISSUE] Make IOManager more configurable
......................................................................
[WIP][NO ISSUE] Make IOManager more configurable
Change-Id: I1c8ad11c2b8b983ef4bf7cf78c2f068accddfff4
---
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
M
hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/controllers/NCConfig.java
M
hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
M
hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/io/IOManager.java
M
hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
M hyracks-fullstack/hyracks/hyracks-storage-common/pom.xml
M
hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
M
hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
M
hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestUtils.java
M
hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/LSMIndexFileManagerTest.java
M
hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/IOManagerPathTest.java
11 files changed, 46 insertions(+), 32 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/33/3133/1
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 d89004b..aced54b 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
@@ -87,6 +87,7 @@
import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
import org.apache.hyracks.api.lifecycle.ILifeCycleComponentManager;
import org.apache.hyracks.api.network.INetworkSecurityManager;
+import org.apache.hyracks.control.common.controllers.NCConfig;
import org.apache.hyracks.control.nc.NodeControllerService;
import org.apache.hyracks.ipc.impl.HyracksConnection;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
@@ -177,6 +178,7 @@
@Override
public void initialize(IRecoveryManagerFactory recoveryManagerFactory,
boolean initialRun) throws IOException {
ioManager = getServiceContext().getIoManager();
+ int ioQueueLen =
getServiceContext().getAppConfig().getInt(NCConfig.Option.IO_QUEUE_SIZE);
threadExecutor =
MaintainedThreadNameExecutorService.newCachedThreadPool(getServiceContext().getThreadFactory());
ICacheMemoryAllocator allocator = new HeapBufferAllocator();
@@ -230,11 +232,11 @@
replicationChannel = new ReplicationChannel(this);
bufferCache = new BufferCache(ioManager, prs, pcp, new
FileMapManager(),
- storageProperties.getBufferCacheMaxOpenFiles(),
getServiceContext().getThreadFactory(),
+ storageProperties.getBufferCacheMaxOpenFiles(),
ioQueueLen, getServiceContext().getThreadFactory(),
replicationManager);
} else {
bufferCache = new BufferCache(ioManager, prs, pcp, new
FileMapManager(),
- storageProperties.getBufferCacheMaxOpenFiles(),
getServiceContext().getThreadFactory());
+ storageProperties.getBufferCacheMaxOpenFiles(),
ioQueueLen, getServiceContext().getThreadFactory());
}
/*
diff --git
a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/controllers/NCConfig.java
b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/controllers/NCConfig.java
index d41350f..19d69f2 100644
---
a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/controllers/NCConfig.java
+++
b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/controllers/NCConfig.java
@@ -89,7 +89,9 @@
TRACE_CATEGORIES(STRING_ARRAY, new String[0]),
KEY_STORE_PATH(STRING, (String) null),
TRUST_STORE_PATH(STRING, (String) null),
- KEY_STORE_PASSWORD(STRING, (String) null);
+ KEY_STORE_PASSWORD(STRING, (String) null),
+ IO_WORKERS_PER_PARTITION(INTEGER, 2),
+ IO_QUEUE_SIZE(INTEGER, 10);
private final IOptionType parser;
private final String defaultValueDescription;
@@ -218,8 +220,12 @@
return "A fully-qualified path to a trust store file that
will be used for secured connections";
case KEY_STORE_PASSWORD:
return "The password to the provided key store";
+ case IO_WORKERS_PER_PARTITION:
+ return "Number of threads per partition used to write and
read from storage";
+ case IO_QUEUE_SIZE:
+ return "Length of the queue used for requests to write and
read";
default:
- throw new IllegalStateException("NYI: " + this);
+ throw new IllegalStateException("Not yet implemented: " +
this);
}
}
@@ -568,4 +574,12 @@
public void setTrustStorePath(String keyStorePath) {
configManager.set(CCConfig.Option.TRUST_STORE_PATH, keyStorePath);
}
+
+ public int getIOParallelism() {
+ return appConfig.getInt(Option.IO_WORKERS_PER_PARTITION);
+ }
+
+ public int getIOQueueLen() {
+ return appConfig.getInt(Option.IO_QUEUE_SIZE);
+ }
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index 317d59a..9fae522 100644
---
a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++
b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -214,8 +214,8 @@
ncShutdownHook = new NCShutdownHook(this);
Runtime.getRuntime().addShutdownHook(ncShutdownHook);
Thread.currentThread().setUncaughtExceptionHandler(getLifeCycleComponentManager());
- ioManager =
- new
IOManager(IODeviceHandle.getDevices(ncConfig.getIODevices()),
application.getFileDeviceResolver());
+ ioManager = new
IOManager(IODeviceHandle.getDevices(ncConfig.getIODevices()),
+ application.getFileDeviceResolver(),
ncConfig.getIOParallelism(), ncConfig.getIOQueueLen());
try {
workQueue = new WorkQueue(id, Thread.NORM_PRIORITY); // Reserves
MAX_PRIORITY of the heartbeat thread.
jobletMap = new ConcurrentHashMap<>();
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 b5cb21a..149d925 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
@@ -56,7 +56,6 @@
/*
* Constants
*/
- public static final int IO_REQUEST_QUEUE_SIZE = 100; // TODO: Make
configurable
private static final Logger LOGGER = LogManager.getLogger();
private static final String WORKSPACE_FILE_SUFFIX = ".waf";
private static final FilenameFilter WORKSPACE_FILES_FILTER = (dir, name)
-> name.endsWith(WORKSPACE_FILE_SUFFIX);
@@ -68,14 +67,19 @@
private final BlockingQueue<IoRequest> freeRequests;
private final List<IODeviceHandle> ioDevices;
private final List<IODeviceHandle> workspaces;
+ private final int ioParallelism;
+ private final int queueLen;
/*
* Mutables
*/
private int workspaceIndex;
private final IFileDeviceResolver deviceComputer;
- public IOManager(List<IODeviceHandle> devices, IFileDeviceResolver
deviceComputer) throws HyracksDataException {
+ public IOManager(List<IODeviceHandle> devices, IFileDeviceResolver
deviceComputer, int ioParallelism, int queueLen)
+ throws HyracksDataException {
this.ioDevices = Collections.unmodifiableList(devices);
+ this.ioParallelism = ioParallelism;
+ this.queueLen = queueLen;
checkDeviceValidity(devices);
workspaces = new ArrayList<>();
for (IODeviceHandle d : ioDevices) {
@@ -93,9 +97,9 @@
}
workspaceIndex = 0;
this.deviceComputer = deviceComputer;
- submittedRequests = new ArrayBlockingQueue<>(IO_REQUEST_QUEUE_SIZE);
- freeRequests = new ArrayBlockingQueue<>(IO_REQUEST_QUEUE_SIZE);
- int numIoThreads = ioDevices.size() * 2;
+ submittedRequests = new ArrayBlockingQueue<>(queueLen);
+ freeRequests = new ArrayBlockingQueue<>(queueLen);
+ int numIoThreads = ioDevices.size() * ioParallelism;
executor = Executors.newFixedThreadPool(numIoThreads);
for (int i = 0; i < numIoThreads; i++) {
executor.execute(new IoRequestHandler(i, submittedRequests));
diff --git
a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
index 32f8418..4005db1 100644
---
a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
+++
b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
@@ -60,7 +60,7 @@
ICacheMemoryAllocator allocator = new HeapBufferAllocator();
IPageReplacementStrategy prs = new
ClockPageReplacementStrategy(allocator, 32768, 50);
bufferCache = new BufferCache(appCtx.getIoManager(), prs, new
DelayPageCleanerPolicy(1000), fileMapManager, 100,
- threadFactory);
+ 10, threadFactory);
ILocalResourceRepositoryFactory localResourceRepositoryFactory = new
TransientLocalResourceRepositoryFactory();
localResourceRepository =
localResourceRepositoryFactory.createRepository();
resourceIdFactory = (new
ResourceIdFactoryProvider(localResourceRepository)).createResourceIdFactory();
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/pom.xml
b/hyracks-fullstack/hyracks/hyracks-storage-common/pom.xml
index 423925b..31037ff 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/pom.xml
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/pom.xml
@@ -50,11 +50,6 @@
</dependency>
<dependency>
<groupId>org.apache.hyracks</groupId>
- <artifactId>hyracks-control-nc</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.hyracks</groupId>
<artifactId>hyracks-util</artifactId>
<version>${project.version}</version>
</dependency>
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 7441395..451e38c 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
@@ -18,8 +18,6 @@
*/
package org.apache.hyracks.storage.common.buffercache;
-import static org.apache.hyracks.control.nc.io.IOManager.IO_REQUEST_QUEUE_SIZE;
-
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
@@ -76,8 +74,7 @@
private final CleanerThread cleanerThread;
private final Map<Integer, BufferedFileHandle> fileInfoMap;
private final AsyncFIFOPageQueueManager fifoWriter;
- private final BlockingQueue<BufferCacheHeaderHelper> headerPageCache =
- new ArrayBlockingQueue<>(IO_REQUEST_QUEUE_SIZE);
+ private final BlockingQueue<BufferCacheHeaderHelper> headerPageCache;
private IIOReplicationManager ioReplicationManager;
private final List<ICachedPageInternal> cachedPages = new ArrayList<>();
@@ -93,8 +90,9 @@
private ConcurrentHashMap<CachedPage, StackTraceElement[]> pinnedPageOwner;
public BufferCache(IIOManager ioManager, IPageReplacementStrategy
pageReplacementStrategy,
- IPageCleanerPolicy pageCleanerPolicy, IFileMapManager
fileMapManager, int maxOpenFiles,
+ IPageCleanerPolicy pageCleanerPolicy, IFileMapManager
fileMapManager, int maxOpenFiles, int ioQueuelen,
ThreadFactory threadFactory) {
+ this.headerPageCache = new ArrayBlockingQueue<>(ioQueuelen);
this.ioManager = ioManager;
this.pageSize = pageReplacementStrategy.getPageSize();
this.maxOpenFiles = maxOpenFiles;
@@ -124,10 +122,11 @@
//this constructor is used when replication is enabled to pass the
IIOReplicationManager
public BufferCache(IIOManager ioManager, IPageReplacementStrategy
pageReplacementStrategy,
- IPageCleanerPolicy pageCleanerPolicy, IFileMapManager
fileMapManager, int maxOpenFiles,
+ IPageCleanerPolicy pageCleanerPolicy, IFileMapManager
fileMapManager, int maxOpenFiles, int ioQueueLen,
ThreadFactory threadFactory, IIOReplicationManager
ioReplicationManager) {
- this(ioManager, pageReplacementStrategy, pageCleanerPolicy,
fileMapManager, maxOpenFiles, threadFactory);
+ this(ioManager, pageReplacementStrategy, pageCleanerPolicy,
fileMapManager, maxOpenFiles, ioQueueLen,
+ threadFactory);
this.ioReplicationManager = ioReplicationManager;
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
index 7b7e850..a2da285 100644
---
a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
+++
b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
@@ -107,7 +107,7 @@
List<IODeviceHandle> devices = new ArrayList<>();
devices.add(new IODeviceHandle(new
File(System.getProperty("user.dir") + File.separator + "target"),
"iodev_test_wa"));
- ioManager = new IOManager(devices, new DefaultDeviceResolver());
+ ioManager = new IOManager(devices, new DefaultDeviceResolver(), 2,
10);
}
return ioManager;
}
@@ -152,7 +152,7 @@
IPageReplacementStrategy prs = new
ClockPageReplacementStrategy(allocator, pageSize, numPages);
IFileMapProvider fileMapProvider = getFileMapProvider();
bufferCache = new BufferCache(ioManager, prs, new
DelayPageCleanerPolicy(1000),
- (IFileMapManager) fileMapProvider, maxOpenFiles,
threadFactory);
+ (IFileMapManager) fileMapProvider, maxOpenFiles, 10,
threadFactory);
return bufferCache;
}
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestUtils.java
b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestUtils.java
index ebfaeb8..e36b655 100644
---
a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestUtils.java
+++
b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestUtils.java
@@ -76,7 +76,7 @@
private static IOManager createIoManager() throws HyracksException {
List<IODeviceHandle> devices = new ArrayList<>();
devices.add(new IODeviceHandle(new
File(System.getProperty("java.io.tmpdir")), "."));
- return new IOManager(devices, new DefaultDeviceResolver());
+ return new IOManager(devices, new DefaultDeviceResolver(), 2, 10);
}
public static void compareWithResult(File expectedFile, File actualFile)
throws Exception {
diff --git
a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/LSMIndexFileManagerTest.java
b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/LSMIndexFileManagerTest.java
index 22456e8..5d10603 100644
---
a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/LSMIndexFileManagerTest.java
+++
b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-common-test/src/test/java/org/apache/hyracks/storage/am/lsm/common/test/LSMIndexFileManagerTest.java
@@ -256,7 +256,7 @@
String iodevPath = System.getProperty("java.io.tmpdir") + sep +
"test_iodev" + i;
devices.add(new IODeviceHandle(new File(iodevPath), "wa"));
}
- return new IOManager(devices, new DefaultDeviceResolver());
+ return new IOManager(devices, new DefaultDeviceResolver(), 2, 10);
}
private FileReference simulateMerge(ILSMIndexFileManager fileManager,
FileReference a, FileReference b)
diff --git
a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/IOManagerPathTest.java
b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/IOManagerPathTest.java
index e2a875b..b0ecc1a 100644
---
a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/IOManagerPathTest.java
+++
b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/IOManagerPathTest.java
@@ -38,8 +38,8 @@
public void testPrefixNames() throws HyracksDataException {
IODeviceHandle shorter = new IODeviceHandle(new File("/tmp/tst/1"),
"storage");
IODeviceHandle longer = new IODeviceHandle(new File("/tmp/tst/11"),
"storage");
- IOManager ioManager =
- new IOManager(Arrays.asList(new IODeviceHandle[] { shorter,
longer }), new DefaultDeviceResolver());
+ IOManager ioManager = new IOManager(Arrays.asList(new IODeviceHandle[]
{ shorter, longer }),
+ new DefaultDeviceResolver(), 2, 10);
FileReference f =
ioManager.resolveAbsolutePath("/tmp/tst/11/storage/Foo_idx_foo/my_btree");
Assert.assertEquals("/tmp/tst/11/storage/Foo_idx_foo/my_btree",
f.getAbsolutePath());
}
@@ -48,8 +48,8 @@
public void testDuplicates() throws HyracksDataException {
IODeviceHandle first = new IODeviceHandle(new File("/tmp/tst/1"),
"storage");
IODeviceHandle second = new IODeviceHandle(new File("/tmp/tst/1"),
"storage");
- IOManager ioManager =
- new IOManager(Arrays.asList(new IODeviceHandle[] { first,
second }), new DefaultDeviceResolver());
+ IOManager ioManager = new IOManager(Arrays.asList(new IODeviceHandle[]
{ first, second }),
+ new DefaultDeviceResolver(), 2, 19);
}
@After
--
To view, visit https://asterix-gerrit.ics.uci.edu/3133
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I1c8ad11c2b8b983ef4bf7cf78c2f068accddfff4
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: Ian Maxon <ima...@apache.org>
