You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@manifoldcf.apache.org by kw...@apache.org on 2011/11/28 01:38:25 UTC
svn commit: r1206939 - in /incubator/lcf/branches/CONNECTORS-286/warthog: ./
src/main/java/org/apache/warthog/common/
src/main/java/org/apache/warthog/interfaces/
src/main/java/org/apache/warthog/tablestore/
src/test/java/org/apache/warthog/tests/
Author: kwright
Date: Mon Nov 28 00:37:54 2011
New Revision: 1206939
URL: http://svn.apache.org/viewvc?rev=1206939&view=rev
Log:
Reduce contention. Also create serialization-only interfaces which are what Voldemort is going to need.
Added:
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java (with props)
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java (with props)
Modified:
incubator/lcf/branches/CONNECTORS-286/warthog/build.xml
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/BooleanValue.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterArray.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterValue.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/IntegerValue.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/LongValue.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringArray.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringValue.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHTableStore.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Index.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Table.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/TableStore.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicKeyValueStore.java
incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/PerformanceTest.java
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/build.xml
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/build.xml?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/build.xml (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/build.xml Mon Nov 28 00:37:54 2011
@@ -76,6 +76,7 @@
<formatter type="brief" usefile="false"/>
<test name="org.apache.warthog.tests.PerformanceTest" todir="build/test-output"/>
+ <test name="org.apache.warthog.tests.MultiThreadTest" todir="build/test-output"/>
<!-- MHL -->
</junit>
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/BooleanValue.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/BooleanValue.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/BooleanValue.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/BooleanValue.java Mon Nov 28 00:37:54 2011
@@ -54,6 +54,18 @@ public class BooleanValue implements WHV
return this.value == ((BooleanValue)value).value;
}
+ public int hashCode()
+ {
+ return value?101:97;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof BooleanValue))
+ return false;
+ return ((BooleanValue)o).value == value;
+ }
+
public static boolean readObject(BufferPointer bp)
{
return ((((int)bp.readValue()) & 0xff) == 0)?false:true;
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterArray.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterArray.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterArray.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterArray.java Mon Nov 28 00:37:54 2011
@@ -53,7 +53,6 @@ public class CharacterArray implements W
return value;
}
-
/** Check if equals (classes must already agree) */
public boolean isEquals(WHValue value)
{
@@ -61,6 +60,31 @@ public class CharacterArray implements W
return compareObject(this.value,otherValue);
}
+ public int hashCode()
+ {
+ int rval = 0;
+ for (int i = 0 ; i < value.length ; i++)
+ {
+ rval += (((int)value[i]) << 5) ^ (((int)value[i]) >> 3);
+ }
+ return rval;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof CharacterArray))
+ return false;
+ CharacterArray sa = (CharacterArray)o;
+ if (sa.value.length != value.length)
+ return false;
+ for (int i = 0 ; i < value.length ; i++)
+ {
+ if (sa.value[i] != value[i])
+ return false;
+ }
+ return true;
+ }
+
public static char[] readObject(BufferPointer bp)
{
int size = IntegerValue.readObject(bp);
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterValue.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterValue.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterValue.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/CharacterValue.java Mon Nov 28 00:37:54 2011
@@ -55,6 +55,18 @@ public class CharacterValue implements W
return value;
}
+ public int hashCode()
+ {
+ return (((int)value) << 5) ^ (((int)value) >> 3);
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof CharacterValue))
+ return false;
+ return ((CharacterValue)o).value == value;
+ }
+
public static char readObject(BufferPointer bp)
{
int value = (((int)bp.readValue()) & 0xff);
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/IntegerValue.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/IntegerValue.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/IntegerValue.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/IntegerValue.java Mon Nov 28 00:37:54 2011
@@ -56,6 +56,18 @@ public class IntegerValue implements WHV
return value;
}
+ public int hashCode()
+ {
+ return (value << 5) ^ (value >> 3);
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof IntegerValue))
+ return false;
+ return ((IntegerValue)o).value == value;
+ }
+
public static int readObject(BufferPointer bp)
{
int value = (((int)bp.readValue()) & 0xff);
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/LongValue.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/LongValue.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/LongValue.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/LongValue.java Mon Nov 28 00:37:54 2011
@@ -56,6 +56,18 @@ public class LongValue implements WHValu
return value;
}
+ public int hashCode()
+ {
+ return (((int)value) << 5) ^ (((int)value) >> 3);
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof LongValue))
+ return false;
+ return ((LongValue)o).value == value;
+ }
+
public static long readObject(BufferPointer bp)
{
long value = (((long)bp.readValue()) & 0xffL);
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringArray.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringArray.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringArray.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringArray.java Mon Nov 28 00:37:54 2011
@@ -56,6 +56,31 @@ public class StringArray implements WHVa
return value;
}
+ public int hashCode()
+ {
+ int rval = 0;
+ for (int i = 0 ; i < value.length ; i++)
+ {
+ rval += value[i].hashCode();
+ }
+ return rval;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof StringArray))
+ return false;
+ StringArray sa = (StringArray)o;
+ if (sa.value.length != value.length)
+ return false;
+ for (int i = 0 ; i < value.length ; i++)
+ {
+ if (!sa.value[i].equals(value[i]))
+ return false;
+ }
+ return true;
+ }
+
public static String[] readObject(BufferPointer bp)
{
int size = IntegerValue.readObject(bp);
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringValue.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringValue.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringValue.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/common/StringValue.java Mon Nov 28 00:37:54 2011
@@ -56,6 +56,18 @@ public class StringValue implements WHVa
return thisValue.value.equals(this.value);
}
+ public int hashCode()
+ {
+ return value.hashCode();
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof StringValue))
+ return false;
+ return ((StringValue)o).value.equals(value);
+ }
+
public static String readObject(BufferPointer bp)
{
char[] array = CharacterArray.readObject(bp);
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,49 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.interfaces;
+import java.util.*;
+
+/** Describes the general functionality requirements of an underlying (distributed?)
+* byte-array-based key value store, which use conflict detection as its mechanism for handling transactional
+* integrity.
+*/
+public interface WHAtomicByteKeyValueStore
+{
+ /** Get a value */
+ public byte[] get(byte[] key)
+ throws WHException;
+
+ /** Check a bunch of values atomically for consistency.
+ *@param checkValues is a map of keys/values that must be unchanged. If any value is
+ * changed, a WHConcurrencyException is thrown.
+ */
+ public void check(WHByteKeyMap checkValues)
+ throws WHException;
+
+ /** Set a bunch of values atomically.
+ *@param checkValues is a map of keys/values that must be unchanged in order for the
+ * commit to proceed. If these values are detected to have been changed, a WHConcurrencyException
+ * will be thrown. Null values are permitted.
+ *@param setValues is a map of keys to set to specified new values. A null value implies removal of
+ * the key.
+ */
+ public void setAll(WHByteKeyMap checkValues, WHByteKeyMap setValues)
+ throws WHException;
+}
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHAtomicByteKeyValueStore.java
------------------------------------------------------------------------------
svn:keywords = Id
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,35 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.interfaces;
+
+/** This interface describes an iterator over a set of byte-based WHKeys. WHException
+* is possible on every iterator step.
+*/
+public interface WHByteKeyIterator
+{
+ /** Check if there is another value */
+ public boolean hasNext()
+ throws WHException;
+
+ /** Get the next value */
+ public byte[] next()
+ throws WHException;
+}
+
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyIterator.java
------------------------------------------------------------------------------
svn:keywords = Id
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,39 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.interfaces;
+
+import java.util.*;
+
+/** This interface describes a set of byte-array-based key/value pairs that
+* is maintained locally.
+*/
+public interface WHByteKeyMap
+{
+ /** Get a value from the map.
+ */
+ public byte[] get(byte[] key)
+ throws WHException;
+
+ /** Iterate over the keys in the map.
+ */
+ public WHByteKeyIterator iterator()
+ throws WHException;
+}
+
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHByteKeyMap.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHTableStore.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHTableStore.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHTableStore.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/interfaces/WHTableStore.java Mon Nov 28 00:37:54 2011
@@ -34,7 +34,7 @@ public interface WHTableStore
public void commitTransaction()
throws WHException;
- /** Abandon a transaction */
+ /** Abandon any transaction in progress */
public void abandonTransaction()
throws WHException;
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Index.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Index.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Index.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Index.java Mon Nov 28 00:37:54 2011
@@ -22,6 +22,8 @@ package org.apache.warthog.tablestore;
import org.apache.warthog.interfaces.*;
import org.apache.warthog.common.*;
+import java.util.*;
+
/** This is the standard implementation of WHIndex.
*
* The index structure has a one-to-one correspondence with the rows
@@ -194,7 +196,9 @@ public class Index implements WHIndex
WHComparator comparator = comparators[columnIndex];
// Read the value we're going to be inserting
WHValue insertValue = table.getValue(rowID,columnName);
- // Keep going until we've found the insertion point
+ // Keep going until we've found the insertion point (or until it looks like we're looping,
+ // which means we've hit a concurrency problem)
+ //Set<LongValue> hitSoFar = new HashSet<LongValue>();
while (true)
{
// Pick up the current node from the parent
@@ -206,6 +210,10 @@ public class Index implements WHIndex
return new IndexNodeColumnKey(indexName,rowID.getValue(),columnName);
}
+ //if (hitSoFar.contains(currentRowID))
+ // throw new WHConcurrencyException();
+ //hitSoFar.add(currentRowID);
+
// Read the value at this node
WHValue currentValue = table.getValue(currentRowID,columnName);
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Table.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Table.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Table.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/Table.java Mon Nov 28 00:37:54 2011
@@ -54,9 +54,8 @@ public class Table implements WHTable
if (columns.length != values.length)
throw new WHException("Column count must agree with value count");
// Allocate the row
- LongValue rowID = allocateNewTableRow();
- // Link in the row
- addTableRow(rowID);
+ LongValue rowID = ts.allocateNewTableRow(tableName);
+ //LongValue rowID = allocateNewTableRow();
// Set the values
for (int i = 0 ; i < columns.length ; i++)
{
@@ -64,6 +63,8 @@ public class Table implements WHTable
}
// Add to all indices
ts.addIndexRow(tableName,rowID,null);
+ // Last, link in the row. Do it last to reduce chances of collision.
+ addTableRow(rowID);
}
/** Update row(s) */
@@ -164,22 +165,48 @@ public class Table implements WHTable
protected void addTableRow(LongValue rowID)
throws WHException
{
- TableTailKey tailKey = new TableTailKey(tableName);
- LongValue tailValue = (LongValue)ts.get(tailKey);
- if (tailValue == null)
+ if (ts.randomDecision())
{
- // First row. Set the head.
- TableHeadKey headKey = new TableHeadKey(tableName);
- ts.put(headKey,rowID);
+ TableTailKey tailKey = new TableTailKey(tableName);
+ LongValue tailValue = (LongValue)ts.get(tailKey);
+ if (tailValue == null)
+ {
+ // First row. Set the head.
+ TableHeadKey headKey = new TableHeadKey(tableName);
+ ts.put(headKey,rowID);
+ }
+ else
+ {
+ // Set the next pointer for last row.
+ TableRowNextKey nextKey = new TableRowNextKey(tableName,tailValue.getValue());
+ ts.put(nextKey,rowID);
+ TableRowPrevKey prevKey = new TableRowPrevKey(tableName,rowID.getValue());
+ ts.put(prevKey,tailValue);
+ }
+ // Set the tail pointer to the new record
+ ts.put(tailKey,rowID);
}
else
{
- // Set the next pointer for last row.
- TableRowNextKey nextKey = new TableRowNextKey(tableName,tailValue.getValue());
- ts.put(nextKey,rowID);
+ TableHeadKey headKey = new TableHeadKey(tableName);
+ LongValue headValue = (LongValue)ts.get(headKey);
+ if (headValue == null)
+ {
+ // First row. Set the tail.
+ TableTailKey tailKey = new TableTailKey(tableName);
+ ts.put(tailKey,rowID);
+ }
+ else
+ {
+ // Set the prev pointer for first row.
+ TableRowPrevKey prevKey = new TableRowPrevKey(tableName,headValue.getValue());
+ ts.put(prevKey,rowID);
+ TableRowNextKey nextKey = new TableRowNextKey(tableName,rowID.getValue());
+ ts.put(nextKey,headValue);
+ }
+ // Set the tail pointer to the new record
+ ts.put(headKey,rowID);
}
- // Set the tail pointer to the new record
- ts.put(tailKey,rowID);
}
/** Remove a table row from the linked list for that table.
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/TableStore.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/TableStore.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/TableStore.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/main/java/org/apache/warthog/tablestore/TableStore.java Mon Nov 28 00:37:54 2011
@@ -40,10 +40,21 @@ public class TableStore implements WHTab
/** Map of table name to index set, good for the current transaction */
protected Map<String,Index[]> indexesPerTable = null;
+ /** Random number generator for this table store instance */
+ protected Random randomGenerator;
+
/** Constructor */
public TableStore(WHTransactionalStore transactionalStore)
{
this.transactionalStore = transactionalStore;
+ randomGenerator = new Random();
+ }
+
+ /** Constructor */
+ public TableStore(WHTransactionalStore transactionalStore, int seed)
+ {
+ this.transactionalStore = transactionalStore;
+ randomGenerator = new Random(seed);
}
/** Begin a transaction */
@@ -71,12 +82,12 @@ public class TableStore implements WHTab
indexesPerTable = null;
}
- /** Abandon a transaction */
+ /** Abandon any transaction in progress */
public void abandonTransaction()
throws WHException
{
if (currentTransaction == null)
- throw new WHException("No transaction to abandon");
+ return;
// As a nicety, tell the transaction we're done.
currentTransaction.abandon();
currentTransaction = null;
@@ -194,6 +205,45 @@ public class TableStore implements WHTab
// Non-interface public methods
+ /** Allocate a new row ID */
+ public LongValue allocateNewTableRow(String tableName)
+ throws WHException
+ {
+ // We always do this in our own transaction to limit the deadlock potential,
+ // and we retry until we get the ID. We only throw a concurrency exception
+ // if it seems clear that the table is gone.
+ TableRowIDFactoryKey tlk = new TableRowIDFactoryKey(tableName);
+
+ while (true)
+ {
+ WHTransaction allocateTransaction = transactionalStore.createTransaction();
+ try
+ {
+ LongValue tlv = (LongValue)allocateTransaction.get(tlk);
+ if (tlv == null)
+ break;
+ allocateTransaction.put(tlk,new LongValue(tlv.getValue()+1L));
+ allocateTransaction.commit();
+ return tlv;
+ }
+ catch (WHConcurrencyException e)
+ {
+ Thread.yield();
+ }
+ finally
+ {
+ allocateTransaction.abandon();
+ }
+ }
+ throw new WHConcurrencyException();
+ }
+
+ /** Random boolean decision */
+ public boolean randomDecision()
+ {
+ return randomGenerator.nextBoolean();
+ }
+
/** Read a value from current transaction.
*/
public WHValue get(WHKey key)
@@ -235,23 +285,21 @@ public class TableStore implements WHTab
indexes[i].addNewRow(rowID);
}
}
-
- protected static boolean indexAffectedBy(Index index, Set<String> columns)
+
+ protected boolean indexAffectedBy(Index index, Set<String> columns)
throws WHException
{
if (columns == null)
return true;
- String[] indexColumns = index.getColumnNames();
- for (int j = 0 ; j < indexColumns.length ; j++)
+ String[] indexColumns = index.getIndexColumns();
+ for (int i = 0 ; i < indexColumns.length ; i++)
{
- if (columns.contains(indexColumns[j]))
+ if (columns.contains(indexColumns[i]))
return true;
}
return false;
}
-
- // Protected classes and methods
-
+
/** Delete a table definition.
* Also deletes associated indexes.
*/
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,109 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.tests;
+
+import org.apache.warthog.interfaces.*;
+import java.util.*;
+
+/** In-memory key value store, for testing.
+*/
+public class InMemAtomicByteKeyValueStore implements WHAtomicByteKeyValueStore
+{
+ protected Map<WrappedByteKey,byte[]> database;
+
+ /** Constructor */
+ public InMemAtomicByteKeyValueStore(int initialSize)
+ {
+ database = new HashMap<WrappedByteKey,byte[]>(initialSize);
+ }
+
+ /** Get a value */
+ public byte[] get(byte[] key)
+ throws WHException
+ {
+ synchronized (database)
+ {
+ return database.get(new WrappedByteKey(key));
+ }
+ }
+
+ /** Check a bunch of values atomically for consistency.
+ *@param checkValues is a map of keys/values that must be unchanged. If any value is
+ * changed, a WHDeadlockException is thrown.
+ */
+ public void check(WHByteKeyMap checkValues)
+ throws WHException
+ {
+ synchronized (database)
+ {
+ WHByteKeyIterator iterator = checkValues.iterator();
+ while (iterator.hasNext())
+ {
+ byte[] key = iterator.next();
+ WrappedByteKey wrappedKey = new WrappedByteKey(key);
+ byte[] value = database.get(wrappedKey);
+ byte[] otherValue = checkValues.get(key);
+ if (value == null && otherValue == null)
+ continue;
+ if (value != null && otherValue != null)
+ {
+ if (value.length != otherValue.length)
+ throw new WHConcurrencyException();
+ for (int i = 0 ; i < value.length ; i++)
+ {
+ if (value[i] != otherValue[i])
+ throw new WHConcurrencyException();
+ }
+ continue;
+ }
+ //System.out.println("Type of key in contention = "+key.getClass().getName());
+ throw new WHConcurrencyException();
+ }
+ }
+ }
+
+ /** Set a bunch of values atomically.
+ *@param checkValues is a map of keys/values that must be unchanged in order for the
+ * commit to proceed. If these values are detected to have been changed, a WHDeadlockException
+ * will be thrown. Null values are permitted.
+ *@param setValues is a map of keys to set to specified new values. A null value implies removal of
+ * the key.
+ */
+ public void setAll(WHByteKeyMap checkValues, WHByteKeyMap setValues)
+ throws WHException
+ {
+ synchronized (database)
+ {
+ check(checkValues);
+ WHByteKeyIterator iterator = setValues.iterator();
+ while (iterator.hasNext())
+ {
+ byte[] key = iterator.next();
+ WrappedByteKey wrappedKey = new WrappedByteKey(key);
+ byte[] value = setValues.get(key);
+ if (value == null)
+ database.remove(wrappedKey);
+ else
+ database.put(wrappedKey,value);
+ }
+ }
+ }
+
+}
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicByteKeyValueStore.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicKeyValueStore.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicKeyValueStore.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicKeyValueStore.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemAtomicKeyValueStore.java Mon Nov 28 00:37:54 2011
@@ -64,6 +64,7 @@ public class InMemAtomicKeyValueStore im
continue;
if (value != null && otherValue != null && value.isEquals(otherValue))
continue;
+ //System.out.println("Type of key in contention = "+key.getClass().getName());
throw new WHConcurrencyException();
}
}
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,232 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.tests;
+
+import org.apache.warthog.interfaces.*;
+import java.util.*;
+
+/** This class implements a byte-based transaction that is in progress, completely in memory.
+*/
+public class InMemByteTransactionImpl implements WHTransaction
+{
+ protected WHAtomicByteKeyValueStore underlyingStore;
+
+ protected KeyMap checkData = new KeyMap();
+ protected KeyMap localData = null;
+
+ /** Constructor */
+ public InMemByteTransactionImpl(WHAtomicByteKeyValueStore underlyingStore)
+ {
+ this.underlyingStore = underlyingStore;
+ }
+
+ /** Set a value. May be null. */
+ public void put(WHKey key, WHValue value)
+ throws WHException
+ {
+ if (localData == null)
+ localData = new KeyMap();
+ localData.put(dematerialize(key),dematerialize(value));
+ }
+
+ /** Get a value. Null returned if no value. */
+ public WHValue get(WHKey key)
+ throws WHException
+ {
+ byte[] byteKey = dematerialize(key);
+ if (localData != null)
+ {
+ if (localData.containsKey(byteKey))
+ return materialize(localData.get(byteKey));
+ }
+ if (checkData.containsKey(byteKey))
+ return materialize(checkData.get(byteKey));
+ byte[] value = underlyingStore.get(byteKey);
+ checkData.put(byteKey,value);
+ return materialize(value);
+ }
+
+ /** Check to see if this transaction has become inconsistent.
+ * If so, a WHDeadlockException is thrown.
+ */
+ public void check()
+ throws WHException
+ {
+ underlyingStore.check(checkData);
+ }
+
+ /** Commit this transaction */
+ public void commit()
+ throws WHException
+ {
+ if (localData != null)
+ underlyingStore.setAll(checkData,localData);
+ else
+ underlyingStore.check(checkData);
+ }
+
+ /** Abandon this transaction.
+ * This is called as a nicety to free any resources associated with the
+ * transaction. The implementation should also be robust as far as
+ * freeing resources if this method is NOT called, but might perform
+ * the necessary logic in a finalizer at an arbitrary time.
+ */
+ public void abandon()
+ {
+ // Does nothing so long as the entire temporary transaction is in
+ // memory.
+ }
+
+ protected byte[] dematerialize(WHValue value)
+ {
+ if (value == null)
+ return null;
+ try
+ {
+ byte[] classNameBytes = value.getClass().getName().getBytes("ASCII");
+ byte[] valueBytes = value.serializeObject();
+ byte[] rval = new byte[1+classNameBytes.length+valueBytes.length];
+ rval[0] = (byte)classNameBytes.length;
+ // Copy arrays around
+ //System.arrayCopy(rval,classNameBytes,1,0,classNameBytes.length);
+ //System.arrayCopy(rval,valueBytes,1+classNameBytes.length,0,valueBytes.length);
+ return rval;
+ }
+ catch (java.io.UnsupportedEncodingException e)
+ {
+ throw new RuntimeException("Couldn't locate ASCII encoder");
+ }
+ }
+
+ protected WHValue materialize(byte[] bytes)
+ throws WHException
+ {
+ if (bytes == null)
+ return null;
+ try
+ {
+ int classNameLength = (int)bytes[0];
+ String className = new String(bytes,1,classNameLength,"ASCII");
+ byte[] serializedValue = new byte[bytes.length - (classNameLength+1)];
+ // Copy bytes around
+ //System.arrayCopy(serializedValue,bytes,0,classNameLength+1,serializedValue.length);
+ try
+ {
+ Class classToMaterialize = Class.forName(className);
+ java.lang.reflect.Constructor constructor = classToMaterialize.getConstructor(new Class[]{byte[].class});
+ Object o = constructor.newInstance(new Object[]{serializedValue});
+ if (!(o instanceof WHValue))
+ throw new WHException("Class '"+className+"' is not an instance of WHValue");
+ return (WHValue)o;
+ }
+ catch (ClassNotFoundException e)
+ {
+ throw new WHException("Class '"+className+"' cannot be found");
+ }
+ catch (java.lang.reflect.InvocationTargetException e)
+ {
+ throw new WHException("Constructor exception instantiating class '"+className+"': "+e.getMessage(),e);
+ }
+ catch (InstantiationException e)
+ {
+ throw new WHException("Instantiation exception for class '"+className+"': "+e.getMessage(),e);
+ }
+ catch (NoSuchMethodException e)
+ {
+ throw new WHException("Class '"+className+"' does not include a constructor with byte[] argument");
+ }
+ catch (IllegalAccessException e)
+ {
+ throw new WHException("Class '"+className+"' byte[] constructor has protected access");
+ }
+ }
+ catch (java.io.UnsupportedEncodingException e)
+ {
+ throw new RuntimeException("Couldn't locate ASCII encoder");
+ }
+ }
+
+ /** Local implementation of byte[] map */
+ protected static class KeyMap implements WHByteKeyMap
+ {
+ protected Map<WrappedByteKey,byte[]> map = new HashMap<WrappedByteKey,byte[]>();
+
+ public KeyMap()
+ {
+ }
+
+ /** Get a value from the map.
+ */
+ public byte[] get(byte[] key)
+ throws WHException
+ {
+ return map.get(new WrappedByteKey(key));
+ }
+
+ /** Iterate over the keys in the map.
+ */
+ public WHByteKeyIterator iterator()
+ throws WHException
+ {
+ return new KeyIterator(map.keySet().iterator());
+ }
+
+ /** Check if the map contains the specified key.
+ */
+ public boolean containsKey(byte[] key)
+ {
+ return map.containsKey(new WrappedByteKey(key));
+ }
+
+ /** Put a value.
+ */
+ public void put(byte[] key, byte[] value)
+ {
+ map.put(new WrappedByteKey(key),value);
+ }
+ }
+
+ /** Key iterator for KeyMap */
+ protected static class KeyIterator implements WHByteKeyIterator
+ {
+ protected Iterator<WrappedByteKey> iterator;
+
+ public KeyIterator(Iterator<WrappedByteKey> iterator)
+ {
+ this.iterator = iterator;
+ }
+
+ /** Check if there is another value */
+ public boolean hasNext()
+ throws WHException
+ {
+ return iterator.hasNext();
+ }
+
+ /** Get the next value */
+ public byte[] next()
+ throws WHException
+ {
+ return iterator.next().getKey();
+ }
+
+ }
+
+}
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionImpl.java
------------------------------------------------------------------------------
svn:keywords = Id
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,44 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.tests;
+
+import org.apache.warthog.interfaces.*;
+
+/** Class implementing a transactional byte-based key/value store completely in memory.
+*/
+public class InMemByteTransactionalStoreImpl implements WHTransactionalStore
+{
+ /** Underlying atomic key/value store */
+ protected WHAtomicByteKeyValueStore underlyingStore;
+
+ /** Constructor. */
+ public InMemByteTransactionalStoreImpl(WHAtomicByteKeyValueStore underlyingStore)
+ {
+ this.underlyingStore = underlyingStore;
+ }
+
+ /** Create a transaction. */
+ public WHTransaction createTransaction()
+ throws WHException
+ {
+ return new InMemByteTransactionImpl(this.underlyingStore);
+ }
+
+}
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/InMemByteTransactionalStoreImpl.java
------------------------------------------------------------------------------
svn:keywords = Id
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,447 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.warthog.tests;
+
+import org.apache.warthog.interfaces.*;
+import org.apache.warthog.common.*;
+import org.apache.warthog.tablestore.*;
+import org.junit.*;
+import static org.junit.Assert.*;
+import java.util.*;
+
+public class MultiThreadTest
+{
+
+ @Test
+ public void writeOnlyTest()
+ throws Exception
+ {
+ int threadCount = 4;
+ int totalInserts = 10000;
+
+ int repeatCount = totalInserts/threadCount;
+
+ WHAtomicKeyValueStore store = new InMemAtomicKeyValueStore(10000000);
+
+ WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(store));
+
+ ts.beginTransaction();
+ ts.createTable("testtable",new String[]{"colA","colB","colC","colD","colE","colF","colG","colH","colI","colJ"});
+ ts.commitTransaction();
+
+ ts.beginTransaction();
+ WHTable table = ts.lookupTable("testtable");
+ ts.createIndex("testindex1",table,
+ new String[]{"colB","colA"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex2",table,
+ new String[]{"colD","colC"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex3",table,
+ new String[]{"colF","colE"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex4",table,
+ new String[]{"colH","colG"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex5",table,
+ new String[]{"colJ","colI"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.commitTransaction();
+
+ InsertThread[] threads = new InsertThread[threadCount];
+ for (int i = 0 ; i < threadCount ; i++)
+ {
+ threads[i] = new InsertThread("Worker thread "+i,store,repeatCount);
+ }
+
+ long startTime = System.currentTimeMillis();
+ for (int i = 0 ; i < threadCount ; i++)
+ {
+ threads[i].start();
+ }
+
+ // Join at the end
+
+ int contentions = 0;
+ for (int i = 0 ; i < threadCount ; i++)
+ {
+ threads[i].join();
+ Throwable e = threads[i].getException();
+ if (e != null)
+ {
+ if (e instanceof Error)
+ throw (Error)e;
+ if (e instanceof RuntimeException)
+ throw (RuntimeException)e;
+ throw (Exception)e;
+ }
+ contentions += threads[i].getContentions();
+ }
+
+ System.out.println(Integer.toString(totalInserts)+" writes across "+
+ Integer.toString(threadCount)+" threads took "+
+ new Long(System.currentTimeMillis()-startTime)+" ms and had "+
+ Integer.toString(contentions)+" contention retries");
+ }
+
+ @Test
+ public void writeQueryTest()
+ throws Exception
+ {
+ int totalCount = 10000/4;
+ int threadCount = 4;
+
+ int repeatCount = totalCount/threadCount;
+
+ WHAtomicKeyValueStore store = new InMemAtomicKeyValueStore(10000000);
+
+ WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(store));
+
+ ts.beginTransaction();
+ ts.createTable("testtable",new String[]{"colA","colB","colC","colD","colE","colF","colG","colH","colI","colJ"});
+ ts.commitTransaction();
+
+ ts.beginTransaction();
+ WHTable table = ts.lookupTable("testtable");
+ ts.createIndex("testindex1",table,
+ new String[]{"colB","colA"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex2",table,
+ new String[]{"colD","colC"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex3",table,
+ new String[]{"colF","colE"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex4",table,
+ new String[]{"colH","colG"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.createIndex("testindex5",table,
+ new String[]{"colJ","colI"},
+ new String[]{"org.apache.warthog.common.StringComparatorAscending",
+ "org.apache.warthog.common.LongComparatorAscending"},false);
+ ts.commitTransaction();
+
+ TaskThread[] threads = new TaskThread[threadCount];
+ for (int i = 0 ; i < threadCount ; i++)
+ {
+ threads[i] = new TaskThread("Worker thread "+i,store,repeatCount);
+ }
+
+ long startTime = System.currentTimeMillis();
+ for (int i = 0 ; i < threadCount ; i++)
+ {
+ threads[i].start();
+ }
+
+ // Join at the end
+
+ int contentions = 0;
+ for (int i = 0 ; i < threadCount ; i++)
+ {
+ threads[i].join();
+ Throwable e = threads[i].getException();
+ if (e != null)
+ {
+ if (e instanceof Error)
+ throw (Error)e;
+ if (e instanceof RuntimeException)
+ throw (RuntimeException)e;
+ throw (Exception)e;
+ }
+ contentions += threads[i].getContentions();
+ }
+
+ System.out.println(Integer.toString(threadCount*repeatCount)+" writes and "+
+ Integer.toString(threadCount*repeatCount*3)+" reads across "+
+ Integer.toString(threadCount)+" threads took "+
+ new Long(System.currentTimeMillis()-startTime)+" ms and had "+
+ Integer.toString(contentions)+" contention retries");
+ }
+
+ // Protected methods
+
+ protected static String randomString(Random r)
+ {
+ return new Long(randomLong(r)).toString();
+ }
+
+ protected static long randomLong(Random r)
+ {
+ return r.nextLong();
+ }
+
+ protected static class InsertThread extends Thread
+ {
+ protected int repeatCount;
+ protected int contentions = 0;
+ protected WHAtomicKeyValueStore store;
+ protected Throwable exception;
+ protected Random r = new Random();
+
+ public InsertThread(String name, WHAtomicKeyValueStore store, int repeatCount)
+ {
+ super();
+ this.repeatCount = repeatCount;
+ this.store = store;
+ setName(name);
+ }
+
+ public void run()
+ {
+ try
+ {
+ WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(store));
+ performInserts(ts);
+ }
+ catch (Throwable e)
+ {
+ exception = e;
+ }
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+
+ public int getContentions()
+ {
+ return contentions;
+ }
+
+ protected void backOff()
+ throws InterruptedException
+ {
+ contentions++;
+ //Thread.sleep(r.nextInt(50));
+ Thread.yield();
+ }
+
+ protected void performInserts(WHTableStore ts)
+ throws InterruptedException, WHException
+ {
+ // Task consists of a certain number of inserts and queries
+ for (int i = 0 ; i < repeatCount ; i++)
+ {
+ // Add a row
+ while (true)
+ {
+ ts.beginTransaction();
+ try
+ {
+ WHTable table = ts.lookupTable("testtable");
+ table.insertRow(new String[]{"colA","colB","colC","colD","colE","colF","colG","colH","colI","colJ"},
+ new WHValue[]{
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r))});
+ ts.commitTransaction();
+ break;
+ }
+ catch (WHConcurrencyException e)
+ {
+ backOff();
+ continue;
+ }
+ finally
+ {
+ ts.abandonTransaction();
+ }
+ }
+ }
+ }
+
+ }
+
+ protected static class TaskThread extends Thread
+ {
+ protected int repeatCount;
+ protected int contentions = 0;
+ protected WHAtomicKeyValueStore store;
+ protected Throwable exception;
+ protected Random r = new Random();
+
+ public TaskThread(String name, WHAtomicKeyValueStore store, int repeatCount)
+ {
+ super();
+ this.repeatCount = repeatCount;
+ this.store = store;
+ setName(name);
+ }
+
+ public void run()
+ {
+ try
+ {
+ WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(store));
+ performTask(ts);
+ }
+ catch (Throwable e)
+ {
+ exception = e;
+ }
+ }
+
+ public Throwable getException()
+ {
+ return exception;
+ }
+
+ public int getContentions()
+ {
+ return contentions;
+ }
+
+ protected void backOff()
+ throws InterruptedException
+ {
+ contentions++;
+ Thread.sleep(r.nextInt(50));
+ //Thread.yield();
+ }
+
+ protected void performTask(WHTableStore ts)
+ throws InterruptedException, WHException
+ {
+ // Task consists of a certain number of inserts and queries
+ for (int i = 0 ; i < repeatCount ; i++)
+ {
+ // Add a row
+ while (true)
+ {
+ ts.beginTransaction();
+ try
+ {
+ WHTable table = ts.lookupTable("testtable");
+ table.insertRow(new String[]{"colA","colB","colC","colD","colE","colF","colG","colH","colI","colJ"},
+ new WHValue[]{
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r)),
+ new LongValue(randomLong(r)),new StringValue(randomString(r))});
+ ts.commitTransaction();
+ break;
+ }
+ catch (WHConcurrencyException e)
+ {
+ backOff();
+ continue;
+ }
+ finally
+ {
+ ts.abandonTransaction();
+ }
+ }
+
+ // Do a query
+ while (true)
+ {
+ ts.beginTransaction();
+ try
+ {
+ WHIndex index = ts.lookupIndex("testindex1");
+ WHAccessor accessor = index.buildAccessor(new IndexCriteria[]{
+ new CriteriaBetween(new StringValue("1"),new StringValue("2")),
+ null});
+ while (accessor.hasNext())
+ accessor.getNext();
+ ts.commitTransaction();
+ break;
+ }
+ catch (WHConcurrencyException e)
+ {
+ backOff();
+ continue;
+ }
+ finally
+ {
+ ts.abandonTransaction();
+ }
+ }
+
+ // Do another query
+ while (true)
+ {
+ ts.beginTransaction();
+ try
+ {
+ WHIndex index = ts.lookupIndex("testindex2");
+ WHAccessor accessor = index.buildAccessor(new IndexCriteria[]{
+ new CriteriaBetween(new StringValue("23"),new StringValue("25")),
+ null});
+ while (accessor.hasNext())
+ accessor.getNext();
+ ts.commitTransaction();
+ break;
+ }
+ catch (WHConcurrencyException e)
+ {
+ backOff();
+ continue;
+ }
+ finally
+ {
+ ts.abandonTransaction();
+ }
+ }
+
+ // Do a third query
+ while (true)
+ {
+ ts.beginTransaction();
+ try
+ {
+ WHIndex index = ts.lookupIndex("testindex3");
+ WHAccessor accessor = index.buildAccessor(new IndexCriteria[]{
+ new CriteriaBetween(new StringValue("02"),new StringValue("04")),
+ null});
+ while (accessor.hasNext())
+ accessor.getNext();
+ ts.commitTransaction();
+ break;
+ }
+ catch (WHConcurrencyException e)
+ {
+ backOff();
+ continue;
+ }
+ finally
+ {
+ ts.abandonTransaction();
+ }
+ }
+
+ }
+ }
+
+ }
+
+}
\ No newline at end of file
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/MultiThreadTest.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/PerformanceTest.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/PerformanceTest.java?rev=1206939&r1=1206938&r2=1206939&view=diff
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/PerformanceTest.java (original)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/PerformanceTest.java Mon Nov 28 00:37:54 2011
@@ -29,7 +29,7 @@ public class PerformanceTest
{
@Test
- public void singleThreadRowCreateTest()
+ public void rowCreateTest()
throws Exception
{
WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(new InMemAtomicKeyValueStore(10000000)));
@@ -104,76 +104,6 @@ public class PerformanceTest
System.out.println("Create of "+Integer.toString(rowCount)+" rows took "+new Long(System.currentTimeMillis() - startTime).toString()+" ms");
}
- @Test
- public void multiThreadWriteQueryTest()
- throws Exception
- {
- WHAtomicKeyValueStore store = new InMemAtomicKeyValueStore(10000000);
-
- WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(store));
-
- ts.beginTransaction();
- ts.createTable("testtable",new String[]{"colA","colB","colC","colD","colE","colF","colG","colH","colI","colJ"});
- ts.commitTransaction();
-
- ts.beginTransaction();
- WHTable table = ts.lookupTable("testtable");
- ts.createIndex("testindex1",table,
- new String[]{"colB","colA"},
- new String[]{"org.apache.warthog.common.StringComparatorAscending",
- "org.apache.warthog.common.LongComparatorAscending"},false);
- ts.createIndex("testindex2",table,
- new String[]{"colD","colC"},
- new String[]{"org.apache.warthog.common.StringComparatorAscending",
- "org.apache.warthog.common.LongComparatorAscending"},false);
- ts.createIndex("testindex3",table,
- new String[]{"colF","colE"},
- new String[]{"org.apache.warthog.common.StringComparatorAscending",
- "org.apache.warthog.common.LongComparatorAscending"},false);
- ts.createIndex("testindex4",table,
- new String[]{"colH","colG"},
- new String[]{"org.apache.warthog.common.StringComparatorAscending",
- "org.apache.warthog.common.LongComparatorAscending"},false);
- ts.createIndex("testindex5",table,
- new String[]{"colJ","colI"},
- new String[]{"org.apache.warthog.common.StringComparatorAscending",
- "org.apache.warthog.common.LongComparatorAscending"},false);
- ts.commitTransaction();
-
- int threadCount = 8;
- TaskThread[] threads = new TaskThread[threadCount];
- for (int i = 0 ; i < threadCount ; i++)
- {
- threads[i] = new TaskThread("Worker thread "+i,store);
- }
-
- long startTime = System.currentTimeMillis();
- for (int i = 0 ; i < threadCount ; i++)
- {
- threads[i].start();
- }
-
- // Join at the end
-
- int contentions = 0;
- for (int i = 0 ; i < threadCount ; i++)
- {
- threads[i].join();
- Throwable e = threads[i].getException();
- if (e != null)
- {
- if (e instanceof Error)
- throw (Error)e;
- if (e instanceof RuntimeException)
- throw (RuntimeException)e;
- throw (Exception)e;
- }
- contentions += threads[i].getContentions();
- }
-
- System.out.println("Task took "+new Long(System.currentTimeMillis()-startTime)+" ms and had "+Integer.toString(contentions)+" contention retries");
- }
-
// Protected methods
protected static String randomString(Random r)
@@ -186,109 +116,4 @@ public class PerformanceTest
return r.nextLong();
}
- protected static class TaskThread extends Thread
- {
- protected int contentions = 0;
- protected WHAtomicKeyValueStore store;
- protected Throwable exception;
- protected Random r = new Random();
-
- public TaskThread(String name, WHAtomicKeyValueStore store)
- {
- super();
- this.store = store;
- setName(name);
- }
-
- public void run()
- {
- try
- {
- WHTableStore ts = new TableStore(new InMemTransactionalStoreImpl(store));
- performTask(ts);
- }
- catch (Throwable e)
- {
- exception = e;
- }
- }
-
- public Throwable getException()
- {
- return exception;
- }
-
- public int getContentions()
- {
- return contentions;
- }
-
- protected void backOff()
- throws InterruptedException
- {
- contentions++;
- Thread.sleep(r.nextInt(200));
- //Thread.yield();
- }
-
- protected void performTask(WHTableStore ts)
- throws InterruptedException, WHException
- {
- // Task consists of a certain number of inserts, deletes, modifies, and queries
- int repeatCount = 1000;
- for (int i = 0 ; i < repeatCount ; i++)
- {
- // Add a row
- while (true)
- {
- ts.beginTransaction();
- try
- {
- WHTable table = ts.lookupTable("testtable");
- table.insertRow(new String[]{"colA","colB","colC","colD","colE","colF","colG","colH","colI","colJ"},
- new WHValue[]{
- new LongValue(randomLong(r)),new StringValue(randomString(r)),
- new LongValue(randomLong(r)),new StringValue(randomString(r)),
- new LongValue(randomLong(r)),new StringValue(randomString(r)),
- new LongValue(randomLong(r)),new StringValue(randomString(r)),
- new LongValue(randomLong(r)),new StringValue(randomString(r))});
- ts.commitTransaction();
- break;
- }
- catch (WHConcurrencyException e)
- {
- ts.abandonTransaction();
- backOff();
- continue;
- }
- }
-
- // Do a query
- while (true)
- {
- ts.beginTransaction();
- try
- {
- WHIndex index = ts.lookupIndex("testindex1");
- WHAccessor accessor = index.buildAccessor(new IndexCriteria[]{
- new CriteriaBetween(new StringValue("1"),new StringValue("2")),
- null});
- while (accessor.hasNext())
- accessor.getNext();
- ts.commitTransaction();
- break;
- }
- catch (WHConcurrencyException e)
- {
- ts.abandonTransaction();
- backOff();
- continue;
- }
- }
-
- }
- }
-
- }
-
}
\ No newline at end of file
Added: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java
URL: http://svn.apache.org/viewvc/incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java?rev=1206939&view=auto
==============================================================================
--- incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java (added)
+++ incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java Mon Nov 28 00:37:54 2011
@@ -0,0 +1,63 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.warthog.tests;
+
+import org.apache.warthog.interfaces.*;
+
+public class WrappedByteKey
+{
+ protected byte[] key;
+
+ public WrappedByteKey(byte[] key)
+ {
+ this.key = key;
+ }
+
+ public int hashCode()
+ {
+ int rval = 0;
+ for (int i = 0 ; i < key.length ; i++)
+ {
+ int current = (int)key[i];
+ rval += (current << 5) ^ (current >> 3);
+ }
+ return rval;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof WrappedByteKey))
+ return false;
+ WrappedByteKey other = (WrappedByteKey)o;
+ if (key.length != other.key.length)
+ return false;
+ for (int i = 0 ; i < key.length ; i++)
+ {
+ if (key[i] != other.key[i])
+ return false;
+ }
+ return true;
+ }
+
+ public byte[] getKey()
+ {
+ return key;
+ }
+}
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/branches/CONNECTORS-286/warthog/src/test/java/org/apache/warthog/tests/WrappedByteKey.java
------------------------------------------------------------------------------
svn:keywords = Id