You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@jena.apache.org by an...@apache.org on 2011/07/28 17:10:12 UTC
svn commit: r1151889 - in /incubator/jena/Experimental/TxTDB/trunk:
src-dev/tx/ src/main/java/com/hp/hpl/jena/tdb/
src/main/java/com/hp/hpl/jena/tdb/transaction/
src/test/java/com/hp/hpl/jena/tdb/transaction/
Author: andy
Date: Thu Jul 28 15:10:10 2011
New Revision: 1151889
URL: http://svn.apache.org/viewvc?rev=1151889&view=rev
Log:
Randomized concurrent testing
Added:
incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/SysTxnState.java (with props)
Modified:
incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/DevTx.java
incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/TxMain.java
incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/StoreConnection.java
incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/TransactionManager.java
incubator/jena/Experimental/TxTDB/trunk/src/test/java/com/hp/hpl/jena/tdb/transaction/TestTransSystem.java
Modified: incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/DevTx.java
URL: http://svn.apache.org/viewvc/incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/DevTx.java?rev=1151889&r1=1151888&r2=1151889&view=diff
==============================================================================
--- incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/DevTx.java (original)
+++ incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/DevTx.java Thu Jul 28 15:10:10 2011
@@ -5,6 +5,7 @@ public class DevTx
{
// --------
// Tasks:
+ // * Check TestTransSystem is actually testing anything.
// * Check journal truncates to last commit.
// Journal needs reset markers
Modified: incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/TxMain.java
URL: http://svn.apache.org/viewvc/incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/TxMain.java?rev=1151889&r1=1151888&r2=1151889&view=diff
==============================================================================
--- incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/TxMain.java (original)
+++ incubator/jena/Experimental/TxTDB/trunk/src-dev/tx/TxMain.java Thu Jul 28 15:10:10 2011
@@ -65,14 +65,6 @@ public class TxMain
public static void main(String... args)
{
- // and one in-mem sConn DB
-
- TestTransSequentialDisk test = new TestTransSequentialDisk() ;
- test.before() ;
- test.trans_readBlock_03() ;
- test.after() ;
- exit(0) ;
-
initFS() ;
String DATA = "/home/afs/Datasets/MusicBrainz/tracks.nt.gz" ;
Modified: incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/StoreConnection.java
URL: http://svn.apache.org/viewvc/incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/StoreConnection.java?rev=1151889&r1=1151888&r2=1151889&view=diff
==============================================================================
--- incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/StoreConnection.java (original)
+++ incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/StoreConnection.java Thu Jul 28 15:10:10 2011
@@ -26,6 +26,7 @@ import com.hp.hpl.jena.tdb.setup.Dataset
import com.hp.hpl.jena.tdb.store.DatasetGraphTDB ;
import com.hp.hpl.jena.tdb.sys.TDBMaker ;
import com.hp.hpl.jena.tdb.transaction.JournalControl ;
+import com.hp.hpl.jena.tdb.transaction.SysTxnState ;
import com.hp.hpl.jena.tdb.transaction.TransactionManager ;
/** Interface to the TDB transaction mechanism. */
@@ -45,6 +46,8 @@ public class StoreConnection
public Location getLocation() { return baseDSG.getLocation() ; }
+ public SysTxnState getTransMgrState() { return transactionManager.state() ; }
+
public DatasetGraphTxn begin(ReadWrite mode)
{
return transactionManager.begin(mode) ;
Added: incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/SysTxnState.java
URL: http://svn.apache.org/viewvc/incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/SysTxnState.java?rev=1151889&view=auto
==============================================================================
--- incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/SysTxnState.java (added)
+++ incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/SysTxnState.java Thu Jul 28 15:10:10 2011
@@ -0,0 +1,52 @@
+/**
+ * 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 com.hp.hpl.jena.tdb.transaction;
+
+public class SysTxnState
+{
+ final public int activeReaders ;
+ final public int activeWriters ;
+ final public int finishedReads ;
+ final public int committedWrite ;
+ final public int abortedWrite ;
+ final public int queuedCommits ;
+
+ SysTxnState(TransactionManager tm)
+ {
+ activeReaders = tm.activeReaders ;
+ activeWriters = tm.activeWriters ;
+ finishedReads = tm.finishedReads ;
+ committedWrite = tm.committedWrite ;
+ abortedWrite = tm.abortedWrite ;
+ queuedCommits = tm.commitedAwaitingFlush.size() ;
+ }
+
+ @Override
+ public String toString()
+ {
+ return String.format("Active (R=%d W=%d) : Finished (R=%d, WC=%d, WA=%d) Queue %d",
+ activeReaders,
+ activeWriters,
+ finishedReads,
+ committedWrite,
+ abortedWrite,
+ queuedCommits
+ ) ;
+ }
+}
Propchange: incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/SysTxnState.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/TransactionManager.java
URL: http://svn.apache.org/viewvc/incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/TransactionManager.java?rev=1151889&r1=1151888&r2=1151889&view=diff
==============================================================================
--- incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/TransactionManager.java (original)
+++ incubator/jena/Experimental/TxTDB/trunk/src/main/java/com/hp/hpl/jena/tdb/transaction/TransactionManager.java Thu Jul 28 15:10:10 2011
@@ -36,38 +36,21 @@ public class TransactionManager
private Set<Transaction> activeTransactions = new HashSet<Transaction>() ;
- // Transactions that have commit (and the journal is written) but haven't
+ // Transactions that have commited (and the journal is written) but haven't
// writted back to the main database.
- private List<Transaction> commitedAwaitingFlush = new ArrayList<Transaction>() ;
+ List<Transaction> commitedAwaitingFlush = new ArrayList<Transaction>() ;
static long transactionId = 1 ;
- private int activeReaders = 0 ;
- private int activeWriters = 0 ; // 0 or 1
+ int activeReaders = 0 ;
+ int activeWriters = 0 ; // 0 or 1
// Misc stats
- private int finishedReads = 0 ;
- private int committedWrite = 0 ;
- private int abortedWrite = 0 ;
-
- public static class State
- {
- final public int activeReaders ;
- final public int activeWriters ;
- final public int finishedReads ;
- final public int committedWrite ;
- final public int abortedWrite ;
- State(TransactionManager tm)
- {
- activeReaders = tm.activeReaders ;
- activeWriters = tm.activeWriters ;
- finishedReads = tm.finishedReads ;
- committedWrite = tm.committedWrite ;
- abortedWrite = tm.abortedWrite ;
- }
- }
-
+ int finishedReads = 0 ;
+ int committedWrite = 0 ;
+ int abortedWrite = 0 ;
+
private BlockingQueue<Transaction> queue = new LinkedBlockingDeque<Transaction>() ;
private Thread committerThread ;
@@ -116,9 +99,9 @@ public class TransactionManager
{
case READ : activeReaders++ ; break ;
case WRITE :
- int x = activeWriters++ ;
- if ( x > 0 )
+ if ( activeWriters > 0 )
throw new TDBTransactionException("Existing active write transaction") ;
+ activeWriters++ ;
break ;
}
@@ -295,8 +278,8 @@ public class TransactionManager
}
synchronized
- public State state()
- { return new State(this) ; }
+ public SysTxnState state()
+ { return new SysTxnState(this) ; }
// LATER.
class Committer implements Runnable
Modified: incubator/jena/Experimental/TxTDB/trunk/src/test/java/com/hp/hpl/jena/tdb/transaction/TestTransSystem.java
URL: http://svn.apache.org/viewvc/incubator/jena/Experimental/TxTDB/trunk/src/test/java/com/hp/hpl/jena/tdb/transaction/TestTransSystem.java?rev=1151889&r1=1151888&r2=1151889&view=diff
==============================================================================
--- incubator/jena/Experimental/TxTDB/trunk/src/test/java/com/hp/hpl/jena/tdb/transaction/TestTransSystem.java (original)
+++ incubator/jena/Experimental/TxTDB/trunk/src/test/java/com/hp/hpl/jena/tdb/transaction/TestTransSystem.java Thu Jul 28 15:10:10 2011
@@ -53,12 +53,15 @@ public class TestTransSystem
static final boolean progress = ! log.isInfoEnabled() ;
- static final int Iterations = 1 ;
- static final int readerSeqRepeats = 10 ;
- static final int readerMaxPause = 50 ;
- static final int writerAbortSeqRepeats = 1 ;
- static final int writerCommitSeqRepeats = 5 ;
- static final int writerMaxPause = 10 ;
+ static final int Iterations = 1 ;
+ static final int numReaderTasks = 20 ;
+ static final int numWriterTasksA = 5 ;
+ static final int numWriterTasksC = 5 ;
+ static final int readerSeqRepeats = 10 ;
+ static final int readerMaxPause = 50 ;
+ static final int writerAbortSeqRepeats = 0 ;
+ static final int writerCommitSeqRepeats = 10 ;
+ static final int writerMaxPause = 20 ;
public static void main(String...args)
{
@@ -150,7 +153,6 @@ public class TestTransSystem
//@Test
public void manyReaderAndOneWriter()
{
- final int numOfTasks = 10 ;
final StoreConnection sConn = getStoreConnection() ;
Callable<?> procR = new Reader(sConn, readerSeqRepeats, readerMaxPause) ; // Number of repeats, max pause
@@ -168,12 +170,10 @@ public class TestTransSystem
{ return changeProc(dsg, id, i) ; }
} ;
- for ( int i = 0 ; i < numOfTasks ; i++ )
- {
- execService.submit(procR) ;
- execService.submit(procW_a) ;
- execService.submit(procW_c) ;
- }
+ submit(execService, procR, numReaderTasks) ;
+ submit(execService, procW_c, numWriterTasksC) ;
+ submit(execService, procW_a, numWriterTasksA) ;
+
try
{
execService.shutdown() ;
@@ -184,6 +184,12 @@ public class TestTransSystem
}
}
+ private void submit(ExecutorService execService2, Callable<?> proc, int numTasks)
+ {
+ for ( int i = 0 ; i < numTasks ; i++ )
+ execService.submit(proc) ;
+ }
+
static int changeProc(DatasetGraphTxn dsg, int id, int i)
{
int count = 0 ;
@@ -214,10 +220,17 @@ public class TestTransSystem
this.maxpause = pause ;
this.sConn = sConn ;
}
-
+
@Override
public Object call()
{
+ try { return call$() ; }
+ catch (RuntimeException ex) { System.err.println(ex.getMessage()) ; return null ; }
+ }
+
+ private Object call$()
+ {
+
int id = gen.incrementAndGet() ;
for ( int i = 0 ; i < repeats; i++ )
{
@@ -228,7 +241,7 @@ public class TestTransSystem
pause(maxpause) ;
int x2 = count("SELECT * { ?s ?p ?o }", dsg) ;
if ( x1 != x2 )
- log.warn(format("Change seen: id=%d: i=%d\n", id, i)) ;
+ log.warn(format("Reader: Change seen: %d/%d : id=%d: i=%d", x1, x2, id, i)) ;
log.debug("reader finish "+id+"/"+i) ;
dsg.close() ;
}
@@ -254,23 +267,33 @@ public class TestTransSystem
@Override
public Object call()
{
+ try { return call$() ; }
+ catch (RuntimeException ex) { System.err.println(ex.getMessage()) ; return null ; }
+ }
+
+ public Object call$()
+ {
int id = gen.incrementAndGet() ;
for ( int i = 0 ; i < repeats ; i++ )
{
+ log.debug("writer start "+id+"/"+i) ;
DatasetGraphTxn dsg = sConn.begin(ReadWrite.WRITE) ;
-System.err.println("writer "+id+"/"+i) ;
int x1 = count("SELECT * { ?s ?p ?o }", dsg) ;
int z = change(dsg, id, i) ;
pause(maxpause) ;
int x2 = count("SELECT * { ?s ?p ?o }", dsg) ;
if ( x1+z != x2 )
- System.err.printf("Change seen: id=%d: i=%d\n", id, i) ;
+ log.warn(format("Writer: Change seen: %d-%d-%d : id=%d: i=%d", x1, z, x2, id, i)) ;
if (commit)
dsg.commit() ;
else
dsg.abort() ;
+ SysTxnState state = sConn.getTransMgrState() ;
+ log.debug(state.toString()) ;
+ log.debug("writer finish "+id+"/"+i) ;
dsg.close() ;
+
}
return null ;
}