10:25 p.m.
Author: shawkins
Date: 2009-08-10 23:25:47 -0400 (Mon, 10 Aug 2009)
New Revision: 1233
Added:
trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java
Removed:
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerFactory.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMap.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMapValueTranslator.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/storage/memory/
trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/memory/
Modified:
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/StorageManager.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/ManagedBatch.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/TupleSourceInfo.java
trunk/engine/src/main/java/com/metamatrix/common/buffer/storage/file/FileStorageManager.java
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBatchMap.java
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBufferManagerImpl.java
trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java
trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java
trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java
trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java
trunk/server/src/main/java/com/metamatrix/server/dqp/service/PlatformBufferService.java
Log:
TEIID-767 merging memorystoremanager directly into the buffermanager.
Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java
===================================================================
--- trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java 2009-08-11
01:26:05 UTC (rev 1232)
+++ trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManager.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -90,12 +90,10 @@
int getConnectorBatchSize();
/**
- * Adds a {@link StorageManager} to this BufferManager instance. This
- * method may be called multiple times; it will be first called after the
- * call to {@link #initialize}.
+ * Adds a {@link StorageManager} to this BufferManager instance.
* @param storageManager Storage manager to add
*/
- void addStorageManager(StorageManager storageManager);
+ void setStorageManager(StorageManager storageManager);
/**
* Creates a tuple source based on a schema and properties describing
Deleted:
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerFactory.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerFactory.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerFactory.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -1,103 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source.
- * See the COPYRIGHT.txt file distributed with this work for information
- * regarding copyright ownership. Some portions may be licensed
- * to Red Hat, Inc. under one or more contributor license agreements.
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
- * 02110-1301 USA.
- */
-
-package com.metamatrix.common.buffer;
-
-import java.util.Properties;
-
-import com.metamatrix.api.exception.MetaMatrixComponentException;
-import com.metamatrix.common.buffer.impl.BufferManagerImpl;
-import com.metamatrix.common.buffer.storage.file.FileStorageManager;
-import com.metamatrix.common.buffer.storage.memory.MemoryStorageManager;
-import com.metamatrix.common.util.PropertiesUtils;
-
-/**
- * <p>Factory for BufferManager instances. One method will get
- * a server buffer manager, as it should be instantiated in a running
- * MetaMatrix server. That BufferManager is configured mostly by the
- * passed in properties.</p>
- *
- * <p>The other method returns a stand-alone, in-memory buffer manager. This
- * is typically used for either in-memory testing or any time the
- * query processor component is not expected to run out of memory, such as
- * within the modeler.</p>
- */
-public class BufferManagerFactory {
-
- private static BufferManager INSTANCE;
-
- /**
- * Helper to get a buffer manager all set up for unmanaged standalone use. This is
- * typically used for testing or when memory is not an issue.
- * @return BufferManager ready for use
- */
- public static BufferManager getStandaloneBufferManager() throws
MetaMatrixComponentException {
- if (INSTANCE == null) {
- BufferManager bufferMgr = new BufferManagerImpl();
- Properties props = new Properties();
- props.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE,
String.valueOf(Long.MAX_VALUE));
- props.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE,
"100"); //$NON-NLS-1$
- props.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL, "0");
//$NON-NLS-1$
- props.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL,
"0"); //$NON-NLS-1$
- bufferMgr.initialize("local", props); //$NON-NLS-1$
-
- // Add unmanaged memory storage manager
- bufferMgr.addStorageManager(new MemoryStorageManager());
- bufferMgr.addStorageManager(new MemoryStorageManager() {
- @Override
- public int getStorageType() {
- return StorageManager.TYPE_FILE;
- }
- });
- INSTANCE = bufferMgr;
- }
-
- return INSTANCE;
- }
-
- /**
- * Helper to get a buffer manager all set up for unmanaged standalone use. This is
- * typically used for testing or when memory is not an issue.
- * @param lookup Lookup implementation to use
- * @param props Configuration properties
- * @return BufferManager ready for use
- */
- public static BufferManager getServerBufferManager(String lookup, Properties props)
throws MetaMatrixComponentException {
- Properties bmProps = PropertiesUtils.clone(props, false);
- // Construct buffer manager
- BufferManager bufferManager = new BufferManagerImpl();
- bufferManager.initialize(lookup, bmProps);
-
- // Get the properties for FileStorageManager and create.
- StorageManager fsm = new FileStorageManager();
- fsm.initialize(bmProps);
- bufferManager.addStorageManager(fsm);
-
- // Create MemoryStorageManager
- StorageManager msm = new MemoryStorageManager();
- msm.initialize(bmProps);
- bufferManager.addStorageManager(msm);
-
- return bufferManager;
- }
-
-}
Modified:
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerPropertyNames.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -28,12 +28,6 @@
public final class BufferManagerPropertyNames {
/**
- * Optional property - the class name of the memory manager to use (must be an
implementation
- * of {@link com.metamatrix.buffer.impl.BufferIDCreator}.
- */
- public static final String ID_CREATOR = "metamatrix.buffer.idCreator";
//$NON-NLS-1$
-
- /**
* Optional property - the amount of memory (in megabytes) that buffer management
should use.
* This value should be a positive integer (less than the the max heap size) and
defaults to
* 128.
Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/StorageManager.java
===================================================================
--- trunk/engine/src/main/java/com/metamatrix/common/buffer/StorageManager.java 2009-08-11
01:26:05 UTC (rev 1232)
+++ trunk/engine/src/main/java/com/metamatrix/common/buffer/StorageManager.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -39,11 +39,6 @@
public interface StorageManager {
/**
- * Constant for a StorageManager for in-memory storage
- */
- public static final int TYPE_MEMORY = 0;
-
- /**
* Constant for a StorageManager for database storage
*/
public static final int TYPE_DATABASE = 1;
@@ -68,16 +63,6 @@
throws MetaMatrixComponentException;
/**
- * Get the type of storage as defined by constants.
- * @return Storage type
- * @see StorageManager#TYPE_MEMORY
- * @see StorageManager#TYPE_DATABASE
- * @see StorageManager#TYPE_FILE
- * @see StorageManager#TYPE_REMOTE
- */
- int getStorageType();
-
- /**
* Add a batch to the storage manager.
* @param types a hint to the StorageManager about the types of data in the batch
* @throws MetaMatrixComponentException indicating a non-business-related
@@ -99,17 +84,6 @@
throws TupleSourceNotFoundException, MetaMatrixComponentException;
/**
- * Remove a batch from this storage as specified. If the tuple source
- * is unknown or the batch is unknown, a TupleSourceNotFoundException is
- * thrown.
- * @throws TupleSourceNotFoundException indicating the sourceID is unknown
- * @throws MetaMatrixComponentException indicating a non-business-related
- * exception (such as a communication exception)
- */
- void removeBatch(TupleSourceID sourceID, int beginRow)
- throws TupleSourceNotFoundException, MetaMatrixComponentException;
-
- /**
* Remove all batches for the specified tuple source. If no batches exist,
* no exception is thrown.
* @throws MetaMatrixComponentException indicating a non-business-related
Deleted: trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMap.java
===================================================================
--- trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMap.java 2009-08-11
01:26:05 UTC (rev 1232)
+++ trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMap.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -1,107 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source.
- * See the COPYRIGHT.txt file distributed with this work for information
- * regarding copyright ownership. Some portions may be licensed
- * to Red Hat, Inc. under one or more contributor license agreements.
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
- * 02110-1301 USA.
- */
-
-package com.metamatrix.common.buffer.impl;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.TreeMap;
-
-
-/**
- * This class is used to save batches ordered by the first row.
- * @since 4.3
- */
-public class BatchMap {
- private TreeMap batches;
- private BatchMapValueTranslator translator;
-
- public BatchMap(BatchMapValueTranslator translator) {
- batches = new TreeMap();
- this.translator = translator;
- }
-
- public void addBatch(int beginRow, Object batch) {
- batches.put(new Integer(beginRow), batch);
- }
-
- public Object getBatch(int beginRow) {
-
- //check special cases first
- int batchSize = batches.size();
- if(batchSize == 0) {
- return null;
- }
- Object batch;
- if(batchSize == 1) {
- batch = batches.values().iterator().next();
- if(beginRow >= translator.getBeginRow(batch) && beginRow <=
translator.getEndRow(batch)) {
- return batch;
- }
- return null;
- }
-
- //Try to search by beginRow. This is fast.
- if((batch = batches.get(new Integer(beginRow))) != null) {
- return batch;
- }
-
- List batchList = new ArrayList(batches.values());
- return doBinarySearchForBatch(batchList, beginRow);
- }
-
- public void removeBatch(int beginRow) {
- if(batches.isEmpty()) {
- return;
- }
- batches.remove(new Integer(beginRow));
- }
-
- public Iterator getBatchIterator() {
- return batches.values().iterator();
- }
-
- public boolean isEmpty() {
- return batches.isEmpty();
- }
-
- private Object doBinarySearchForBatch(List batches, int beginRow) {
- int batchSize = batches.size();
- int beginIndex = 0;
- int midIndex;
- int endIndex = batchSize;
- Object batch;
- while(beginIndex < endIndex) {
- midIndex = (beginIndex + endIndex)/2;
- batch = batches.get(midIndex);
- if(beginRow < translator.getBeginRow(batch)){
- endIndex = midIndex;
- }else if(beginRow > translator.getEndRow(batch)) {
- beginIndex = midIndex + 1;
- }else {
- return batch;
- }
- }
- return null;
- }
-}
Deleted:
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMapValueTranslator.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMapValueTranslator.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BatchMapValueTranslator.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -1,45 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source.
- * See the COPYRIGHT.txt file distributed with this work for information
- * regarding copyright ownership. Some portions may be licensed
- * to Red Hat, Inc. under one or more contributor license agreements.
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
- * 02110-1301 USA.
- */
-
-package com.metamatrix.common.buffer.impl;
-
-
-/**
- * @since 4.3
- */
-public interface BatchMapValueTranslator {
- /**
- * Return the beginning row number of the given batch.
- * @param batchMapValue
- * @return
- * @since 4.3
- */
- int getBeginRow(Object batchMapValue);
-
- /**
- * Return the ending row number of the given batch.
- * @param batchMapValue
- * @return
- * @since 4.3
- */
- int getEndRow(Object batchMapValue);
-}
Modified:
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/BufferManagerImpl.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -77,7 +77,6 @@
private Map<String, TupleGroupInfo> groupInfos = new HashMap<String,
TupleGroupInfo>();
// Storage managers
- private StorageManager memoryMgr;
private StorageManager diskMgr;
// ID creator
@@ -174,9 +173,9 @@
}
synchronized(info) {
- Iterator batchIter = info.getBatchIterator();
+ Iterator<ManagedBatch> batchIter = info.getBatchIterator();
while(batchIter.hasNext()) {
- ManagedBatch batch = (ManagedBatch) batchIter.next();
+ ManagedBatch batch = batchIter.next();
switch(batch.getLocation()) {
case ManagedBatch.PERSISTENT:
stats.numPersistentBatches++;
@@ -226,14 +225,10 @@
* Add a storage manager to this buffer manager, order is unimportant
* @param storageManager Storage manager to add
*/
- public void addStorageManager(StorageManager storageManager) {
+ public void setStorageManager(StorageManager storageManager) {
Assertion.isNotNull(storageManager);
- if(storageManager.getStorageType() == StorageManager.TYPE_MEMORY) {
- this.memoryMgr = storageManager;
- } else {
- Assertion.isNull(diskMgr);
- this.diskMgr = storageManager;
- }
+ Assertion.isNull(diskMgr);
+ this.diskMgr = storageManager;
}
/**
@@ -289,9 +284,9 @@
if(! info.isRemoved()) {
info.setRemoved();
- Iterator iter = info.getBatchIterator();
+ Iterator<ManagedBatch> iter = info.getBatchIterator();
while(iter.hasNext()) {
- ManagedBatch batch = (ManagedBatch) iter.next();
+ ManagedBatch batch = iter.next();
switch(batch.getLocation()) {
case ManagedBatch.UNPINNED:
memoryState.removeUnpinned(batch);
@@ -309,9 +304,7 @@
synchronized (tupleGroupInfo) {
tupleGroupInfo.getTupleSourceIDs().remove(tupleSourceID);
}
- // Remove memory storage
- this.memoryMgr.removeBatches(tupleSourceID);
-
+
// Remove disk storage
if (this.diskMgr != null){
this.diskMgr.removeBatches(tupleSourceID);
@@ -491,12 +484,16 @@
throw new
TupleSourceNotFoundException(QueryExecPlugin.Util.getString("BufferManagerImpl.tuple_source_not_found",
tupleSourceID)); //$NON-NLS-1$
}
+ // Update tuple source state
+ ManagedBatch managedBatch = new ManagedBatch(tupleSourceID,
tupleBatch.getBeginRow(), tupleBatch.getEndRow(), bytes);
+ managedBatch.setLocation(location);
+
// Store into storage manager
try {
if(location == ManagedBatch.PERSISTENT) {
this.diskMgr.addBatch(tupleSourceID, tupleBatch, info.getTypes());
} else {
- this.memoryMgr.addBatch(tupleSourceID, tupleBatch, info.getTypes());
+ managedBatch.setBatch(tupleBatch);
}
} catch(MetaMatrixComponentException e) {
// If we were storing to memory, clean up memory we reserved
@@ -505,11 +502,7 @@
}
throw e;
}
-
- // Update tuple source state
- ManagedBatch managedBatch = new ManagedBatch(tupleSourceID,
tupleBatch.getBeginRow(), tupleBatch.getEndRow(), bytes);
- managedBatch.setLocation(location);
-
+
// Add to memory state if in memory
if(location == ManagedBatch.UNPINNED) {
this.memoryState.addUnpinned(managedBatch);
@@ -565,7 +558,7 @@
} else if(mbatch.getLocation() == ManagedBatch.PINNED) {
// Load batch from memory - already pinned
- memoryBatch = this.memoryMgr.getBatch(tupleSourceID, beginRow,
info.getTypes());
+ memoryBatch = mbatch.getBatch();
} else if(mbatch.getLocation() == ManagedBatch.UNPINNED) {
// Already in memory - just move from unpinned to pinned
@@ -574,7 +567,7 @@
this.memoryState.addPinned(mbatch);
// Load batch from memory
- memoryBatch = this.memoryMgr.getBatch(tupleSourceID, beginRow,
info.getTypes());
+ memoryBatch = mbatch.getBatch();
} else if(mbatch.getLocation() == ManagedBatch.PERSISTENT) {
memoryRequiredByBatch = mbatch.getSize();
@@ -601,26 +594,7 @@
int internalBeginRow = mbatch.getBeginRow();
memoryBatch = diskMgr.getBatch(tupleSourceID, internalBeginRow,
info.getTypes());
- try {
- memoryMgr.addBatch(tupleSourceID, memoryBatch, info.getTypes());
- } catch(MetaMatrixComponentException e) {
- memoryState.releaseMemory(mbatch.getSize(),
info.getGroupInfo());
- throw e;
- }
-
- try {
- diskMgr.removeBatch(tupleSourceID, internalBeginRow);
- } catch(TupleSourceNotFoundException e) {
- } catch(MetaMatrixComponentException e) {
- memoryState.releaseMemory(memoryRequiredByBatch,
info.getGroupInfo());
- try {
- memoryMgr.removeBatch(tupleSourceID, internalBeginRow);
- } catch(Exception e2) {
- // ignore
- }
- throw e;
- }
-
+ mbatch.setBatch(memoryBatch);
mbatch.setLocation(ManagedBatch.PINNED);
this.memoryState.addPinned(mbatch);
}
@@ -736,13 +710,13 @@
long totalMemory = config.getTotalAvailableMemory();
long released = 0;
- Iterator unpinnedIter = this.memoryState.getAllUnpinned();
+ Iterator<ManagedBatch> unpinnedIter = this.memoryState.getAllUnpinned();
while(unpinnedIter.hasNext() && // If there are unpinned batches in
memory, AND
// Defect 14573 - if we require more than what's available, then
cleanup regardless of the threshold
(memoryRequired > totalMemory - memoryState.getMemoryUsed() || // if the
memory needed is more than what's available, or
memoryState.getMemoryUsed() > targetLevel)){ // if we've crossed
the active memory threshold, then cleanup
- ManagedBatch batch = (ManagedBatch) unpinnedIter.next();
+ ManagedBatch batch = unpinnedIter.next();
TupleSourceID tsID = batch.getTupleSourceID();
released += releaseMemory(batch, tsID);
@@ -763,9 +737,9 @@
boolean cleanForSessionSucceeded = false;
long released = 0;
- Iterator unpinnedIter = this.memoryState.getAllUnpinned();
+ Iterator<ManagedBatch> unpinnedIter = this.memoryState.getAllUnpinned();
while(unpinnedIter.hasNext()) {
- ManagedBatch batch = (ManagedBatch) unpinnedIter.next();
+ ManagedBatch batch = unpinnedIter.next();
TupleSourceID tsID = batch.getTupleSourceID();
TupleSourceInfo tsInfo = getTupleSourceInfo(tsID, false);
if(tsInfo == null) {
@@ -828,15 +802,7 @@
}
// This batch is still unpinned - move to persistent storage
- int beginRow = batch.getBeginRow();
- TupleBatch dataBatch = null;
- try {
- dataBatch = memoryMgr.getBatch(tsID, beginRow, info.getTypes());
- } catch(TupleSourceNotFoundException e) {
- return 0;
- } catch(MetaMatrixComponentException e) {
- return 0;
- }
+ TupleBatch dataBatch = batch.getBatch();
try {
diskMgr.addBatch(tsID, dataBatch, info.getTypes());
@@ -845,13 +811,7 @@
return 0;
}
- try {
- memoryMgr.removeBatch(tsID, beginRow);
- } catch(TupleSourceNotFoundException e) {
- // ignore
- } catch(MetaMatrixComponentException e) {
- // ignore
- }
+ batch.setBatch(null);
// Update memory
batch.setLocation(ManagedBatch.PERSISTENT);
@@ -1048,18 +1008,18 @@
* @see com.metamatrix.common.buffer.BufferManager#releasePinnedBatches()
*/
public void releasePinnedBatches() throws MetaMatrixComponentException {
- Map threadPinned = memoryState.getPinnedByCurrentThread();
+ Map<TupleSourceID, Map<Integer, ManagedBatch>> threadPinned =
memoryState.getPinnedByCurrentThread();
if (threadPinned == null) {
return;
}
- for (Iterator i = threadPinned.entrySet().iterator(); i.hasNext();) {
- Map.Entry entry = (Map.Entry)i.next();
+ for (Iterator<Map.Entry<TupleSourceID, Map<Integer,
ManagedBatch>>> i = threadPinned.entrySet().iterator(); i.hasNext();) {
+ Map.Entry<TupleSourceID, Map<Integer, ManagedBatch>> entry =
i.next();
i.remove();
- TupleSourceID tsid = (TupleSourceID)entry.getKey();
- Map pinnedBatches = (Map)entry.getValue();
+ TupleSourceID tsid = entry.getKey();
+ Map<Integer, ManagedBatch> pinnedBatches = entry.getValue();
try {
- for (Iterator j = pinnedBatches.values().iterator(); j.hasNext();) {
- ManagedBatch batch = (ManagedBatch)j.next();
+ for (Iterator<ManagedBatch> j = pinnedBatches.values().iterator();
j.hasNext();) {
+ ManagedBatch batch = j.next();
//TODO: add trace logging about the batch that is being unpinned
unpinTupleBatch(tsid, batch.getBeginRow(), batch.getEndRow());
@@ -1074,7 +1034,7 @@
* for testing purposes
*/
public int getPinnedCount() {
- Map pinned = memoryState.getAllPinned();
+ Map<TupleSourceID, Map<Integer, ManagedBatch>> pinned =
memoryState.getAllPinned();
int count = 0;
@@ -1082,8 +1042,8 @@
return count;
}
- for (Iterator i = pinned.values().iterator(); i.hasNext();) {
- count += ((Map)i.next()).size();
+ for (Iterator<Map<Integer, ManagedBatch>> i =
pinned.values().iterator(); i.hasNext();) {
+ count += i.next().size();
}
return count;
Modified: trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/ManagedBatch.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/ManagedBatch.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/ManagedBatch.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -26,6 +26,7 @@
import java.util.LinkedList;
import java.util.List;
+import com.metamatrix.common.buffer.TupleBatch;
import com.metamatrix.common.buffer.TupleSourceID;
import com.metamatrix.core.util.HashCodeUtil;
@@ -56,12 +57,13 @@
private int location;
private int pinnedCount;
private int hashCode;
+ private TupleBatch batch;
// logging
private static int STACK_LEVELS_TO_OMIT = 2;
private static int STACK_LEVELS_TO_CAPTURE = 5;
- private List whoCalledUs;
+ private List<String> whoCalledUs;
private String sCallStackTimeStamp;
/**
@@ -75,6 +77,14 @@
this.hashCode = HashCodeUtil.hashCode(tupleSourceID.hashCode(), beginRow);
}
+ public TupleBatch getBatch() {
+ return batch;
+ }
+
+ public void setBatch(TupleBatch batch) {
+ this.batch = batch;
+ }
+
/**
* Get the tuple source ID
* @return Tuple sourceID
@@ -147,7 +157,7 @@
*/
StackTraceElement[] elements = new Exception().getStackTrace();
- whoCalledUs = new LinkedList();
+ whoCalledUs = new LinkedList<String>();
for ( int i = STACK_LEVELS_TO_OMIT; i < elements.length && i <
STACK_LEVELS_TO_OMIT + STACK_LEVELS_TO_CAPTURE; i++ ) {
whoCalledUs.add(elements[ i ].toString());
@@ -158,7 +168,7 @@
/**
* Returns call stack
*/
- public List getCallStack() {
+ public List<String> getCallStack() {
return whoCalledUs;
}
Modified:
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/TupleSourceInfo.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/TupleSourceInfo.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/impl/TupleSourceInfo.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -34,15 +34,6 @@
*/
public class TupleSourceInfo {
- private static final BatchMapValueTranslator TRANSLATOR = new
BatchMapValueTranslator() {
- public int getBeginRow(Object batchMapValue) {
- return ((ManagedBatch)batchMapValue).getBeginRow();
- }
- public int getEndRow(Object batchMapValue) {
- return ((ManagedBatch)batchMapValue).getEndRow();
- }
- };
-
private TupleSourceType type; // Type of TupleSource, as defined in
BufferManager constants
private TupleSourceID tsID;
private List schema;
@@ -51,8 +42,8 @@
private TupleSourceStatus status;
private TupleGroupInfo groupInfo;
private boolean removed = false;
+ private TreeMap<Integer, ManagedBatch> batches = new TreeMap<Integer,
ManagedBatch>();
- private BatchMap batches;
private boolean lobs;
/**
@@ -63,7 +54,6 @@
* @param type Type of tuple source, as defined in BufferManager constants
*/
public TupleSourceInfo(TupleSourceID tsID, List schema, String[] types,
TupleGroupInfo groupInfo, TupleSourceType type) {
-
this.tsID = tsID;
this.schema = schema;
this.types = types;
@@ -71,10 +61,29 @@
this.status = TupleSourceStatus.ACTIVE;
this.rowCount = 0;
this.type = type;
- this.batches = new BatchMap(TRANSLATOR);
this.lobs = checkForLobs();
}
+ public void addBatch(ManagedBatch batch) {
+ batches.put(batch.getBeginRow(), batch);
+ }
+
+ public ManagedBatch getBatch(int beginRow) {
+ Map.Entry<Integer, ManagedBatch> entry = batches.floorEntry(beginRow);
+ if (entry != null && entry.getValue().getEndRow() >= beginRow) {
+ return entry.getValue();
+ }
+ return null;
+ }
+
+ public void removeBatch(int beginRow) {
+ batches.remove(new Integer(beginRow));
+ }
+
+ public Iterator<ManagedBatch> getBatchIterator() {
+ return batches.values().iterator();
+ }
+
/**
* Get the tuple source identifier
* @return Tuple source identifier
@@ -159,47 +168,10 @@
this.removed = true;
}
- /**
- * Add a managed batch to this tuple source
- * @param batch Managed batch for this tuple source
- */
- public void addBatch(ManagedBatch batch) {
- batches.addBatch(batch.getBeginRow(), batch);
- }
-
- /**
- * Remove a managed batch for this tuple source,
- * indexed by beginRow
- * @param beginRow First row of batch to remove
- */
- public void removeBatch(int beginRow) {
- batches.removeBatch(beginRow);
- }
-
- /**
- * Get a batch, specified by beginRow
- * @param beginRow First row of batch to retrieve
- * @return Managed batch or null if not found
- */
- public ManagedBatch getBatch(int beginRow) {
- return (ManagedBatch)batches.getBatch(beginRow);
- }
-
public boolean lobsInSource() {
return this.lobs;
}
- /**
- * Get iterator directly on the batches contained by this
- * tuple source. To use this safely, the user should ensure
- * through locking that no other thread are modifying this
- * TupleSourceInfo while the iterator is being used.
- * @return Unsafe iterator
- */
- public Iterator getBatchIterator() {
- return this.batches.getBatchIterator();
- }
-
public String[] getTypes() {
return types;
}
Modified:
trunk/engine/src/main/java/com/metamatrix/common/buffer/storage/file/FileStorageManager.java
===================================================================
---
trunk/engine/src/main/java/com/metamatrix/common/buffer/storage/file/FileStorageManager.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/main/java/com/metamatrix/common/buffer/storage/file/FileStorageManager.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -33,8 +33,8 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -72,7 +72,7 @@
private File dirFile;
// State
- private Map tupleSourceMap = new HashMap(); // TupleSourceID ->
TupleSourceInfo
+ private Map<TupleSourceID, TupleSourceInfo> tupleSourceMap = new
HashMap<TupleSourceID, TupleSourceInfo>();
private Map<File, RandomAccessFile> fileCache = Collections.synchronizedMap(new
LinkedHashMap<File, RandomAccessFile>() {
@Override
protected boolean removeEldestEntry(
@@ -119,14 +119,6 @@
}
/**
- * Return file type: {@link com.metamatrix.common.buffer.StorageManager.TYPE_FILE}
- * @return File type constant
- */
- public int getStorageType() {
- return StorageManager.TYPE_FILE;
- }
-
- /**
* Look up tuple source info and possibly create. First the file map is used to find
an
* existing file info. If the info is found it is returned. If not, then
* a TupleSourceInfo is created according to shouldCreate flag
@@ -138,7 +130,7 @@
// Try to find in cache
synchronized(tupleSourceMap) {
- TupleSourceInfo info = (TupleSourceInfo) tupleSourceMap.get(sourceID);
+ TupleSourceInfo info = tupleSourceMap.get(sourceID);
if(info == null && shouldCreate) {
info = new TupleSourceInfo();
tupleSourceMap.put(sourceID, info);
@@ -285,7 +277,7 @@
throw new
TupleSourceNotFoundException(QueryExecPlugin.Util.getString("BufferManagerImpl.tuple_source_not_found",
sourceID)); //$NON-NLS-1$
}
// Find pointer
- PointerInfo pointerInfo = (PointerInfo) info.tupleBatchPointers.get(new
Integer(beginRow));
+ PointerInfo pointerInfo = info.tupleBatchPointers.get(new
Integer(beginRow));
Assertion.isNotNull(pointerInfo);
FileInfo fileInfo = pointerInfo.fileInfo;
@@ -305,9 +297,9 @@
batch.readExternal(ois);
return batch;
} catch(IOException e) {
- throw new MetaMatrixComponentException(e,
QueryExecPlugin.Util.getString("FileStoreageManager.error_reading",
fileInfo.file.getAbsoluteFile()));
+ throw new MetaMatrixComponentException(e,
QueryExecPlugin.Util.getString("FileStoreageManager.error_reading",
fileInfo.file.getAbsoluteFile())); //$NON-NLS-1$
} catch (ClassNotFoundException e) {
- throw new MetaMatrixComponentException(e,
QueryExecPlugin.Util.getString("FileStoreageManager.error_reading",
fileInfo.file.getAbsoluteFile()));
+ throw new MetaMatrixComponentException(e,
QueryExecPlugin.Util.getString("FileStoreageManager.error_reading",
fileInfo.file.getAbsoluteFile())); //$NON-NLS-1$
} finally {
fileInfo.close();
}
@@ -315,19 +307,6 @@
}
/**
- * This method does nothing - rather than deleting batches from the middle of a
RandomAccessFile,
- * which would be very expensive, we just handle the possibility that a batch already
exists
- * in the addBatch method.
- * @param sourceID Source identifier
- * @param beginRow Beginning batch row to remove
- */
- public void removeBatch(TupleSourceID sourceID, int beginRow)
- throws TupleSourceNotFoundException, MetaMatrixComponentException {
-
- // nothing - don't remove batches as it is too expensive
- }
-
- /**
* Remove all batches for a sourceID. Before removal, the file is closed.
* @param sourceID Tuple source ID
*/
@@ -335,7 +314,7 @@
TupleSourceInfo info = null;
// Remove info from the file map
synchronized(tupleSourceMap) {
- info = (TupleSourceInfo)tupleSourceMap.remove(sourceID);
+ info = tupleSourceMap.remove(sourceID);
}
// Didn't find a file
@@ -355,7 +334,7 @@
// If open, close the file and decrement the open file counter
for (int i = 0; i < info.storageFiles.size(); i++) {
- FileInfo fileInfo = (FileInfo)info.storageFiles.get(i);
+ FileInfo fileInfo = info.storageFiles.get(i);
fileInfo.delete();
}
// Delete the file and mark info as being removed
@@ -371,10 +350,12 @@
LogManager.logDetail(LogConstants.CTX_STORAGE_MGR, "Removing all storage area
files "); //$NON-NLS-1$
- Iterator tsIter = tupleSourceMap.keySet().iterator();
-
- while(tsIter.hasNext()) {
- TupleSourceID key = (TupleSourceID)tsIter.next();
+ List<TupleSourceID> ids = null;
+ synchronized (tupleSourceMap) {
+ ids = new LinkedList<TupleSourceID>(tupleSourceMap.keySet());
+ }
+
+ for (TupleSourceID key : ids) {
try {
removeBatches(key);
} catch (MetaMatrixComponentException e) {
@@ -398,10 +379,6 @@
this.file = file;
}
- public boolean isOpen() {
- return fileData != null;
- }
-
public void open() throws FileNotFoundException {
if(this.fileData == null) {
this.fileData = fileCache.remove(this.file);
@@ -454,15 +431,15 @@
}
private static class TupleSourceInfo {
- Map tupleBatchPointers = new HashMap(); // beginRow -> PointerInfo
- List storageFiles = new ArrayList(2); // Stores all the FileInfos for this
tupleSource
+ Map<Integer, PointerInfo> tupleBatchPointers = new HashMap<Integer,
PointerInfo>();
+ List<FileInfo> storageFiles = new ArrayList<FileInfo>(2); // Stores
all the FileInfos for this tupleSource
private boolean isRemoved = false;
FileInfo getMostRecentlyCreatedFile() {
if (storageFiles.isEmpty()) {
return null;
}
- return (FileInfo)storageFiles.get(storageFiles.size() - 1);
+ return storageFiles.get(storageFiles.size() - 1);
}
}
Copied: trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java
(from rev 1214,
trunk/engine/src/main/java/com/metamatrix/common/buffer/BufferManagerFactory.java)
===================================================================
--- trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java
(rev 0)
+++
trunk/engine/src/test/java/com/metamatrix/common/buffer/BufferManagerFactory.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -0,0 +1,141 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * See the COPYRIGHT.txt file distributed with this work for information
+ * regarding copyright ownership. Some portions may be licensed
+ * to Red Hat, Inc. under one or more contributor license agreements.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301 USA.
+ */
+
+package com.metamatrix.common.buffer;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import com.metamatrix.api.exception.MetaMatrixComponentException;
+import com.metamatrix.common.buffer.impl.BufferManagerImpl;
+import com.metamatrix.query.execution.QueryExecPlugin;
+
+/**
+ * <p>Factory for BufferManager instances. One method will get
+ * a server buffer manager, as it should be instantiated in a running
+ * MetaMatrix server. That BufferManager is configured mostly by the
+ * passed in properties.</p>
+ *
+ * <p>The other method returns a stand-alone, in-memory buffer manager. This
+ * is typically used for either in-memory testing or any time the
+ * query processor component is not expected to run out of memory, such as
+ * within the modeler.</p>
+ */
+public class BufferManagerFactory {
+
+ public static class MemoryStorageManager implements StorageManager {
+
+ // TupleSourceID -> List<TupleBatch> (ordered by startRow)
+ private Map<TupleSourceID, Map<Integer, TupleBatch>> storage =
Collections.synchronizedMap(new HashMap<TupleSourceID, Map<Integer,
TupleBatch>>());
+
+ /**
+ * @see StorageManager#initialize(Properties)
+ */
+ public void initialize(Properties props) throws MetaMatrixComponentException {
+ }
+
+ /**
+ * @see StorageManager#addBatch(TupleSourceID, TupleBatch)
+ */
+ public void addBatch(TupleSourceID storageID, TupleBatch batch, String[] types)
+ throws MetaMatrixComponentException {
+
+ Map<Integer, TupleBatch> batches = null;
+ synchronized(this.storage) {
+ batches = storage.get(storageID);
+ if(batches == null) {
+ batches = new HashMap<Integer, TupleBatch>();
+ this.storage.put(storageID, batches);
+ }
+ }
+
+ synchronized(batches) {
+ batches.put(batch.getBeginRow(), batch);
+ }
+ }
+
+ /**
+ * @see StorageManager#getBatch(TupleSourceID, int, int)
+ */
+ public TupleBatch getBatch(TupleSourceID storageID, int beginRow, String[] types)
+ throws TupleSourceNotFoundException, MetaMatrixComponentException {
+
+ Map<Integer, TupleBatch> batches = storage.get(storageID);
+
+ if(batches == null) {
+ throw new
TupleSourceNotFoundException(QueryExecPlugin.Util.getString("BufferManagerImpl.tuple_source_not_found",
storageID)); //$NON-NLS-1$
+ }
+
+ synchronized(batches) {
+ TupleBatch batch = batches.get(beginRow);
+ if(batch == null) {
+ throw new MetaMatrixComponentException("unknown batch");
//$NON-NLS-1$
+ }
+ return batch;
+ }
+ }
+
+ /**
+ * @see StorageManager#removeStorageArea(TupleSourceID)
+ */
+ public void removeBatches(TupleSourceID storageID) throws
MetaMatrixComponentException {
+ storage.remove(storageID);
+ }
+
+ /**
+ * @see StorageManager#shutdown()
+ */
+ public void shutdown() {
+ this.storage.clear();
+ this.storage = null;
+ }
+
+ }
+
+ private static BufferManager INSTANCE;
+
+ /**
+ * Helper to get a buffer manager all set up for unmanaged standalone use. This is
+ * typically used for testing or when memory is not an issue.
+ * @return BufferManager ready for use
+ */
+ public static BufferManager getStandaloneBufferManager() throws
MetaMatrixComponentException {
+ if (INSTANCE == null) {
+ BufferManager bufferMgr = new BufferManagerImpl();
+ Properties props = new Properties();
+ props.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE,
String.valueOf(Long.MAX_VALUE));
+ props.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE,
"100"); //$NON-NLS-1$
+ props.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL, "0");
//$NON-NLS-1$
+ props.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL,
"0"); //$NON-NLS-1$
+ bufferMgr.initialize("local", props); //$NON-NLS-1$
+
+ // Add unmanaged memory storage manager
+ bufferMgr.setStorageManager(new MemoryStorageManager());
+ INSTANCE = bufferMgr;
+ }
+
+ return INSTANCE;
+ }
+
+}
Modified: trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBatchMap.java
===================================================================
---
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBatchMap.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBatchMap.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -28,6 +28,8 @@
import java.util.List;
import java.util.Set;
+import com.metamatrix.common.buffer.TupleSourceID;
+
import junit.framework.TestCase;
@@ -35,14 +37,6 @@
* @since 4.3
*/
public class TestBatchMap extends TestCase{
- private BatchMapValueTranslator batchMapValueTranslator = new
BatchMapValueTranslator() {
- public int getBeginRow(Object batchMapValue) {
- return ((Batch)batchMapValue).getBeginRow();
- }
- public int getEndRow(Object batchMapValue) {
- return ((Batch)batchMapValue).getEndRow();
- }
- };
public TestBatchMap(String arg0) {
super(arg0);
@@ -50,17 +44,16 @@
//no batch
public void testAddAndGetBatch1() {
- BatchMap batches = new BatchMap(batchMapValueTranslator);
- assertTrue(batches.isEmpty());
+ TupleSourceInfo batches = new TupleSourceInfo(null, null, null, null, null);
assertNull(batches.getBatch(1));
assertNull(batches.getBatch(2));
}
//one batch
public void testAddAndGetBatch2() {
- BatchMap batches = new BatchMap(batchMapValueTranslator);
- Batch batch1 = new Batch(1,5);
- batches.addBatch(1, batch1);
+ TupleSourceInfo batches = new TupleSourceInfo(null, null, null, null, null);
+ ManagedBatch batch1 = createManagedBatch(1, 5);
+ batches.addBatch(batch1);
assertEquals(batch1, batches.getBatch(1));
assertEquals(batch1, batches.getBatch(2));
assertEquals(batch1, batches.getBatch(5));
@@ -70,11 +63,11 @@
//two batches
public void testAddAndGetBatch3() {
- BatchMap batches = new BatchMap(batchMapValueTranslator);
- Batch batch1 = new Batch(1,5);
- Batch batch2 = new Batch(6,10);
- batches.addBatch(1, batch1);
- batches.addBatch(6, batch2);
+ TupleSourceInfo batches = new TupleSourceInfo(null, null, null, null, null);
+ ManagedBatch batch1 = createManagedBatch(1, 5);
+ ManagedBatch batch2 = createManagedBatch(6, 10);
+ batches.addBatch(batch1);
+ batches.addBatch(batch2);
assertEquals(batch1, batches.getBatch(1));
assertEquals(batch1, batches.getBatch(2));
assertEquals(batch1, batches.getBatch(5));
@@ -83,22 +76,26 @@
assertEquals(batch2, batches.getBatch(10));
assertNull(batches.getBatch(11));
}
+
+ private ManagedBatch createManagedBatch(int begin, int end) {
+ return new ManagedBatch(new TupleSourceID("x"), begin, end, 0);
//$NON-NLS-1$
+ }
//more batches, no adding order
public void testAddAndGetBatch4() {
- BatchMap batches = new BatchMap(batchMapValueTranslator);
+ TupleSourceInfo batches = new TupleSourceInfo(null, null, null, null, null);
Set batchSet = new HashSet();
List batchList = new ArrayList();
for(int i=1; i<10000;) {
- Batch batch = new Batch(i,i+4);
+ ManagedBatch batch = createManagedBatch(i, i + 4);
batchSet.add(batch);
batchList.add(batch);
i += 5;
}
Iterator iter = batchSet.iterator();
while(iter.hasNext()) {
- Batch next = (Batch)iter.next();
- batches.addBatch(next.getBeginRow(), next);
+ ManagedBatch next = (ManagedBatch)iter.next();
+ batches.addBatch(next);
}
for(int i=1; i<10000;) {
assertEquals(batchList.get(i/5), batches.getBatch(i));
@@ -112,39 +109,23 @@
}
public void testAddAndRemoveBatch() {
- BatchMap batches = new BatchMap(batchMapValueTranslator);
- assertTrue(batches.isEmpty());
- Batch batch1 = new Batch(1,5);
- Batch batch2 = new Batch(6,10);
- batches.addBatch(1, batch1);
- batches.addBatch(2, batch2);
+ TupleSourceInfo batches = new TupleSourceInfo(null, null, null, null, null);
+ ManagedBatch batch1 = createManagedBatch(1, 5);
+ ManagedBatch batch2 = createManagedBatch(6, 10);
+ batches.addBatch(batch1);
+ batches.addBatch(batch2);
assertEquals(batch1, batches.getBatch(1));
batches.removeBatch(1);
assertNull(batches.getBatch(1));
assertNull(batches.getBatch(2));
assertNull(batches.getBatch(5));
- batches.removeBatch(2);
+ batches.removeBatch(6);
assertNull(batches.getBatch(1));
assertNull(batches.getBatch(2));
assertNull(batches.getBatch(5));
assertNull(batches.getBatch(6));
assertNull(batches.getBatch(8));
assertNull(batches.getBatch(10));
- assertTrue(batches.isEmpty());
}
- class Batch{
- private int beginRow;
- private int endRow;
- Batch(int beginRow, int endRow){
- Batch.this.beginRow = beginRow;
- Batch.this.endRow = endRow;
- }
- int getBeginRow() {
- return Batch.this.beginRow;
- }
- int getEndRow() {
- return Batch.this.endRow;
- }
- }
}
Modified:
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBufferManagerImpl.java
===================================================================
---
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBufferManagerImpl.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/test/java/com/metamatrix/common/buffer/impl/TestBufferManagerImpl.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -47,7 +47,7 @@
import com.metamatrix.common.buffer.TupleSourceNotFoundException;
import com.metamatrix.common.buffer.BufferManager.TupleSourceStatus;
import com.metamatrix.common.buffer.BufferManager.TupleSourceType;
-import com.metamatrix.common.buffer.storage.memory.MemoryStorageManager;
+import com.metamatrix.common.buffer.BufferManagerFactory.MemoryStorageManager;
import com.metamatrix.common.lob.ByteLobChunkStream;
import com.metamatrix.common.lob.LobChunk;
import com.metamatrix.common.lob.LobChunkInputStream;
@@ -69,7 +69,7 @@
super(arg0);
}
- public static BufferManager getTestBufferManager(long bytesAvailable, StorageManager
sm1, StorageManager sm2) throws MetaMatrixComponentException {
+ public static BufferManager getTestBufferManager(long bytesAvailable, StorageManager
sm2) throws MetaMatrixComponentException {
// Get the properties for BufferManager
Properties bmProps = new Properties();
bmProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" +
bytesAvailable); //$NON-NLS-1$
@@ -77,30 +77,19 @@
BufferManager bufferManager = new BufferManagerImpl();
bufferManager.initialize("local", bmProps); //$NON-NLS-1$
- // Add storage managers
- bufferManager.addStorageManager(sm1);
-
if(sm2 != null) {
- bufferManager.addStorageManager(sm2);
+ bufferManager.setStorageManager(sm2);
}
return bufferManager;
}
- private StorageManager createMemoryStorageManager() {
- return new MemoryStorageManager();
- }
-
private StorageManager createFakeDatabaseStorageManager() {
- return new MemoryStorageManager() {
- public int getStorageType() {
- return StorageManager.TYPE_DATABASE;
- }
- };
+ return new MemoryStorageManager();
}
- public void helpTestAddBatches(StorageManager sm1, StorageManager sm2, int
memorySize, int numBatches, int rowsPerBatch) throws MetaMatrixComponentException,
MetaMatrixProcessingException {
- BufferManager mgr = getTestBufferManager(memorySize, sm1, sm2);
+ public void helpTestAddBatches(StorageManager sm2, int memorySize, int numBatches,
int rowsPerBatch) throws MetaMatrixComponentException, MetaMatrixProcessingException {
+ BufferManager mgr = getTestBufferManager(memorySize, sm2);
List expectedRows = new ArrayList();
@@ -136,20 +125,28 @@
// Check batches in sm1 - 3 should fit in 1000 bytes of memory
int batchesInMemory = (int) (memorySize * 1000000 / batchSize);
- for(int b=0; b<batchesInMemory; b++) {
- int begin = (b*rowsPerBatch) + 1;
-
- TupleBatch batch = sm1.getBatch(tsID, begin, null);
- assertNotNull("batch " + (b+1) + " is null ", batch);
//$NON-NLS-1$ //$NON-NLS-2$
- }
+ if (sm2 != null) {
+ for(int b=0; b<batchesInMemory; b++) {
+ int begin = (b*rowsPerBatch) + 1;
+ try {
+ sm2.getBatch(tsID, begin, null);
+ fail("Expected exception"); //$NON-NLS-1$
+ } catch (TupleSourceNotFoundException e) {
+
+ } catch (MetaMatrixComponentException e) {
+
+ }
+ }
+
+ // Check batches in sm2
+ for(int b=batchesInMemory+1; b<numBatches; b++) {
+ int begin = (b*rowsPerBatch) + 1;
+
+ TupleBatch batch = sm2.getBatch(tsID, begin, null);
+ assertNotNull("batch " + (b+1) + " is null ", batch);
//$NON-NLS-1$ //$NON-NLS-2$
+ }
+ }
- // Check batches in sm2
- for(int b=batchesInMemory+1; b<numBatches; b++) {
- int begin = (b*rowsPerBatch) + 1;
-
- TupleBatch batch = sm2.getBatch(tsID, begin, null);
- assertNotNull("batch " + (b+1) + " is null ", batch);
//$NON-NLS-1$ //$NON-NLS-2$
- }
// Check row count
int rowCount = mgr.getRowCount(tsID);
@@ -168,16 +165,14 @@
}
public void testSpanStorage() throws Exception {
- helpTestAddBatches(createMemoryStorageManager(),
- createFakeDatabaseStorageManager(),
+ helpTestAddBatches(createFakeDatabaseStorageManager(),
1,
50,
100);
}
public void testStandalone() throws Exception {
- helpTestAddBatches(createMemoryStorageManager(),
- null,
+ helpTestAddBatches(null,
10,
5,
10);
@@ -218,7 +213,7 @@
}
public void testCreateLobReference() throws Exception {
- final BufferManagerImpl mgr = (BufferManagerImpl)getTestBufferManager(1,
createMemoryStorageManager(), createFakeDatabaseStorageManager());
+ final BufferManagerImpl mgr = (BufferManagerImpl)getTestBufferManager(1,
createFakeDatabaseStorageManager());
XMLType xml1 = new XMLType(new SQLXMLImpl("<foo/>"));
//$NON-NLS-1$
XMLType xml2 = new XMLType(new SQLXMLImpl("<bar/>"));
//$NON-NLS-1$
@@ -257,7 +252,7 @@
}
public void testAddStreamablePart() throws Exception {
- final BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ final BufferManager mgr = getTestBufferManager(1,
createFakeDatabaseStorageManager());
// save the lob
List schema = new ArrayList();
@@ -318,7 +313,7 @@
public void testPinning1() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
List schema = new ArrayList();
schema.add("val"); //$NON-NLS-1$
@@ -349,7 +344,7 @@
}
public void testUnpinOfUnpinnedBatch() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
List schema = new ArrayList();
schema.add("val"); //$NON-NLS-1$
@@ -403,7 +398,7 @@
}
public void testDeadlockOnMultiThreadClean() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
List schema = new ArrayList();
schema.add("val"); //$NON-NLS-1$
@@ -440,7 +435,7 @@
}
public void testSessionMax_Fail() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
TupleSourceID tsID = null;
try {
List schema = new ArrayList();
@@ -481,7 +476,7 @@
* @since 4.3
*/
public void testDefect_18499() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
List schema = new ArrayList();
schema.add("col"); //$NON-NLS-1$
String[] schemaTypes = new String[] {DataTypeManager.DefaultDataTypes.STRING};
@@ -502,7 +497,7 @@
}
public void testDefect18497() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
TupleSourceID tsID = null;
try {
List schema = new ArrayList();
@@ -536,7 +531,7 @@
//two threads do the cleaning at the same time
public void testDefect19325() throws Exception{
- BufferManagerImpl mgr = (BufferManagerImpl)getTestBufferManager(1,
createMemoryStorageManager(), createFakeDatabaseStorageManager());
+ BufferManagerImpl mgr = (BufferManagerImpl)getTestBufferManager(1,
createFakeDatabaseStorageManager());
TupleSourceID tsID = null;
List schema = new ArrayList();
schema.add("val"); //$NON-NLS-1$
@@ -572,7 +567,7 @@
//test many small batches
public void testSmallBatches() throws Exception{
- BufferManager mgr = getTestBufferManager(50, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(50,
createFakeDatabaseStorageManager());
TupleSourceID tsID = null;
try {
List schema = new ArrayList();
@@ -606,7 +601,7 @@
//going backward
public void testPinning2() throws Exception {
- BufferManager mgr = getTestBufferManager(1, createMemoryStorageManager(),
createFakeDatabaseStorageManager());
+ BufferManager mgr = getTestBufferManager(1, createFakeDatabaseStorageManager());
List schema = new ArrayList();
schema.add("val"); //$NON-NLS-1$
Modified:
trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java
===================================================================
---
trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/test/java/com/metamatrix/common/buffer/storage/file/TestFileStorageManager.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -322,7 +322,7 @@
}
}
- public void testAddTwice() {
+ public void testAddTwice() throws Exception {
StorageManager sm = getStorageManager(null);
TupleSourceID tsID = new TupleSourceID("local,1:0"); //$NON-NLS-1$
TupleBatch batch = exampleBatch(1, 20);
@@ -331,14 +331,9 @@
// Add batch
sm.addBatch(tsID, batch, null);
- // Remove batch (does nothing)
- sm.removeBatch(tsID, batch.getBeginRow());
-
// Add batch again
sm.addBatch(tsID, batch, null);
- } catch(MetaMatrixException e) {
- fail("Unexpected exception of type " + e.getClass().getName() +
": " + e.getMessage()); //$NON-NLS-1$ //$NON-NLS-2$
} finally {
try {
sm.removeBatches(tsID);
@@ -527,7 +522,6 @@
// Remove
if(added[batch]) {
//System.out.println(tsID.toString() + ": removing batch
" + batch);
- sm.removeBatch(tsID, batches[batch].getBeginRow());
added[batch] = false;
}
}
Modified:
trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java
===================================================================
---
trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/test/java/com/metamatrix/query/processor/TestVirtualDepJoin.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -38,8 +38,8 @@
import com.metamatrix.common.buffer.BufferManager;
import com.metamatrix.common.buffer.BufferManagerPropertyNames;
import com.metamatrix.common.buffer.TupleSourceNotFoundException;
+import com.metamatrix.common.buffer.BufferManagerFactory.MemoryStorageManager;
import com.metamatrix.common.buffer.impl.BufferManagerImpl;
-import com.metamatrix.common.buffer.storage.memory.MemoryStorageManager;
import com.metamatrix.common.types.DataTypeManager;
import com.metamatrix.query.mapping.relational.QueryNode;
import com.metamatrix.query.optimizer.TestOptimizer;
@@ -491,17 +491,16 @@
private BufferManager createCustomBufferMgr(int batchSize) throws
MetaMatrixComponentException {
BufferManager bufferMgr = new BufferManagerImpl();
Properties props = new Properties();
- props.setProperty(BufferManagerPropertyNames.ID_CREATOR,
"com.metamatrix.common.buffer.impl.LongIDCreator"); //$NON-NLS-1$
- props.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" +
Long.MAX_VALUE); //$NON-NLS-1$
+ props.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE,
String.valueOf(Long.MAX_VALUE));
props.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE,
"100"); //$NON-NLS-1$
props.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL, "0");
//$NON-NLS-1$
props.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL, "0");
//$NON-NLS-1$
- props.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, "" +
batchSize); //$NON-NLS-1$
- props.setProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE, "" +
batchSize);//$NON-NLS-1$
+ props.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE,
String.valueOf(batchSize));
+ props.setProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE,
String.valueOf(batchSize));
bufferMgr.initialize("local", props); //$NON-NLS-1$
// Add unmanaged memory storage manager
- bufferMgr.addStorageManager(new MemoryStorageManager());
+ bufferMgr.setStorageManager(new MemoryStorageManager());
return bufferMgr;
}
Modified:
trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java
===================================================================
---
trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/engine/src/test/java/com/metamatrix/query/processor/relational/NodeTestUtil.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -33,8 +33,8 @@
import com.metamatrix.common.buffer.TupleBatch;
import com.metamatrix.common.buffer.TupleSourceID;
import com.metamatrix.common.buffer.TupleSourceNotFoundException;
+import com.metamatrix.common.buffer.BufferManagerFactory.MemoryStorageManager;
import com.metamatrix.common.buffer.impl.BufferManagerImpl;
-import com.metamatrix.common.buffer.storage.memory.MemoryStorageManager;
import com.metamatrix.core.MetaMatrixRuntimeException;
@@ -45,7 +45,6 @@
static BufferManager getTestBufferManager(long bytesAvailable) {
// Get the properties for BufferManager
Properties bmProps = new Properties();
- bmProps.setProperty(BufferManagerPropertyNames.ID_CREATOR,
"com.metamatrix.common.buffer.impl.LongIDCreator"); //$NON-NLS-1$
bmProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" +
bytesAvailable); //$NON-NLS-1$
return createBufferManager(bmProps);
}
@@ -54,7 +53,6 @@
// Get the properties for BufferManager
Properties bmProps = new Properties();
- bmProps.setProperty(BufferManagerPropertyNames.ID_CREATOR,
"com.metamatrix.common.buffer.impl.LongIDCreator"); //$NON-NLS-1$
bmProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" +
bytesAvailable); //$NON-NLS-1$
bmProps.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, ""
+ procBatchSize); //$NON-NLS-1$
bmProps.setProperty(BufferManagerPropertyNames.CONNECTOR_BATCH_SIZE, ""
+ connectorBatchSize); //$NON-NLS-1$
@@ -65,7 +63,6 @@
// Get the properties for BufferManager
Properties bmProps = new Properties();
- bmProps.setProperty(BufferManagerPropertyNames.ID_CREATOR,
"com.metamatrix.common.buffer.impl.LongIDCreator"); //$NON-NLS-1$
bmProps.setProperty(BufferManagerPropertyNames.MEMORY_AVAILABLE, "" +
bytesAvailable); //$NON-NLS-1$
bmProps.setProperty(BufferManagerPropertyNames.PROCESSOR_BATCH_SIZE, ""
+ procBatchSize); //$NON-NLS-1$
return createBufferManager(bmProps);
@@ -81,22 +78,14 @@
}
// Add storage managers
- bufferManager.addStorageManager(createMemoryStorageManager());
- bufferManager.addStorageManager(createFakeDatabaseStorageManager());
+ bufferManager.setStorageManager(createFakeDatabaseStorageManager());
return bufferManager;
}
- private static StorageManager createMemoryStorageManager() {
- return new MemoryStorageManager();
- }
private static StorageManager createFakeDatabaseStorageManager() {
- return new MemoryStorageManager() {
- public int getStorageType() {
- return StorageManager.TYPE_DATABASE;
- }
- };
+ return new MemoryStorageManager();
}
public static class TestableBufferManagerImpl extends BufferManagerImpl {
Modified:
trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java
===================================================================
---
trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/runtime/src/main/java/com/metamatrix/dqp/embedded/services/EmbeddedBufferService.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -38,7 +38,6 @@
import com.metamatrix.common.buffer.StorageManager;
import com.metamatrix.common.buffer.impl.BufferManagerImpl;
import com.metamatrix.common.buffer.storage.file.FileStorageManager;
-import com.metamatrix.common.buffer.storage.memory.MemoryStorageManager;
import com.metamatrix.dqp.embedded.DQPEmbeddedPlugin;
import com.metamatrix.dqp.service.BufferService;
import com.metamatrix.dqp.service.ConfigurationService;
@@ -102,7 +101,6 @@
// Set up buffer configuration properties
Properties bufferProps = new Properties();
- bufferProps.setProperty(BufferManagerPropertyNames.ID_CREATOR,
DEFAULT_ID_CREATOR);
bufferProps.setProperty(BufferManagerPropertyNames.SESSION_USE_PERCENTAGE,
DEFAULT_SESSION_USE_PERCENTAGE);
bufferProps.setProperty(BufferManagerPropertyNames.LOG_STATS_INTERVAL,
DEFAULT_LOG_STATS_INTERVAL);
bufferProps.setProperty(BufferManagerPropertyNames.MANAGEMENT_INTERVAL,
DEFAULT_MANAGEMENT_INTERVAL);
@@ -126,16 +124,13 @@
fsmProps.setProperty(BufferManagerPropertyNames.MAX_OPEN_FILES,
DEFAULT_MAX_OPEN_FILES);
StorageManager fsm = new FileStorageManager();
fsm.initialize(fsmProps);
- this.bufferMgr.addStorageManager(fsm);
+ this.bufferMgr.setStorageManager(fsm);
// start the file storrage manager in clean state
// wise FileStorageManager is smart enough to clen up after itself
cleanDirectory(bufferDir);
}
- // Add unmanaged memory storage manager
- this.bufferMgr.addStorageManager(new MemoryStorageManager());
-
} catch(MetaMatrixComponentException e) {
throw new ApplicationLifecycleException(e,
DQPEmbeddedPlugin.Util.getString("LocalBufferService.Failed_initializing_buffer_manager._8"));
//$NON-NLS-1$
} catch(IOException e) {
Modified:
trunk/server/src/main/java/com/metamatrix/server/dqp/service/PlatformBufferService.java
===================================================================
---
trunk/server/src/main/java/com/metamatrix/server/dqp/service/PlatformBufferService.java 2009-08-11
01:26:05 UTC (rev 1232)
+++
trunk/server/src/main/java/com/metamatrix/server/dqp/service/PlatformBufferService.java 2009-08-11
03:25:47 UTC (rev 1233)
@@ -33,8 +33,10 @@
import com.metamatrix.common.application.exception.ApplicationInitializationException;
import com.metamatrix.common.application.exception.ApplicationLifecycleException;
import com.metamatrix.common.buffer.BufferManager;
-import com.metamatrix.common.buffer.BufferManagerFactory;
import com.metamatrix.common.buffer.BufferManagerPropertyNames;
+import com.metamatrix.common.buffer.StorageManager;
+import com.metamatrix.common.buffer.impl.BufferManagerImpl;
+import com.metamatrix.common.buffer.storage.file.FileStorageManager;
import com.metamatrix.common.config.api.Host;
import com.metamatrix.common.util.PropertiesUtils;
import com.metamatrix.core.util.FileUtils;
@@ -87,12 +89,33 @@
props.setProperty(BufferManagerPropertyNames.BUFFER_STORAGE_DIRECTORY, dir);
try {
- bufferMgr =
BufferManagerFactory.getServerBufferManager(host.getFullName()+"-"+processName,
props); //$NON-NLS-1$
+ bufferMgr = getServerBufferManager(host.getFullName()+"-"+processName,
props); //$NON-NLS-1$
} catch (MetaMatrixComponentException e) {
throw new ApplicationLifecycleException(e);
}
}
+
+ /**
+ * Helper to get a buffer manager all set up for unmanaged standalone use. This is
+ * typically used for testing or when memory is not an issue.
+ * @param lookup Lookup implementation to use
+ * @param props Configuration properties
+ * @return BufferManager ready for use
+ */
+ public static BufferManager getServerBufferManager(String lookup, Properties props)
throws MetaMatrixComponentException {
+ Properties bmProps = PropertiesUtils.clone(props, false);
+ // Construct buffer manager
+ BufferManager bufferManager = new BufferManagerImpl();
+ bufferManager.initialize(lookup, bmProps);
+ // Get the properties for FileStorageManager and create.
+ StorageManager fsm = new FileStorageManager();
+ fsm.initialize(bmProps);
+ bufferManager.setStorageManager(fsm);
+
+ return bufferManager;
+ }
+
/*
* @see com.metamatrix.common.application.ApplicationService#stop()
*/