Remove duplicate ThreadTxn from DBOE. Project: http://git-wip-us.apache.org/repos/asf/jena/repo Commit: http://git-wip-us.apache.org/repos/asf/jena/commit/74808450 Tree: http://git-wip-us.apache.org/repos/asf/jena/tree/74808450 Diff: http://git-wip-us.apache.org/repos/asf/jena/diff/74808450
Branch: refs/heads/master Commit: 74808450e635418cca758a564288fc80994b3ad6 Parents: 0366da2 Author: Andy Seaborne <[email protected]> Authored: Fri Dec 29 18:56:57 2017 +0000 Committer: Andy Seaborne <[email protected]> Committed: Sun Dec 31 10:53:24 2017 +0000 ---------------------------------------------------------------------- .../java/org/apache/jena/system/ThreadTxn.java | 1 - .../trans/data/TestTransBinaryDataFile.java | 5 +- .../jena/dboe/trans/data/TestTransBlob.java | 5 +- .../apache/jena/dboe/transaction/ThreadTxn.java | 144 ------------------- .../transaction/TestThreadingTransactions.java | 17 +-- .../TestTransactionCoordinatorControl.java | 8 +- .../jena/dboe/transaction/TestTxnLib.java | 15 +- .../jena/dboe/transaction/TestTxnSwitching.java | 9 +- 8 files changed, 31 insertions(+), 173 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-arq/src/main/java/org/apache/jena/system/ThreadTxn.java ---------------------------------------------------------------------- diff --git a/jena-arq/src/main/java/org/apache/jena/system/ThreadTxn.java b/jena-arq/src/main/java/org/apache/jena/system/ThreadTxn.java index 4b2fc46..0b73821 100644 --- a/jena-arq/src/main/java/org/apache/jena/system/ThreadTxn.java +++ b/jena-arq/src/main/java/org/apache/jena/system/ThreadTxn.java @@ -32,7 +32,6 @@ import org.apache.jena.sparql.core.Transactional ; * will cause deadlock. */ public class ThreadTxn { - // ---- Thread /** Create a thread-backed delayed READ transaction action. * Call {@link ThreadAction#run} to perform the read transaction. http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBinaryDataFile.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBinaryDataFile.java b/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBinaryDataFile.java index d30f75d..996749b 100644 --- a/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBinaryDataFile.java +++ b/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBinaryDataFile.java @@ -27,16 +27,15 @@ import org.apache.jena.atlas.lib.StrUtils ; import org.apache.jena.dboe.base.file.*; import org.apache.jena.dboe.jenax.Txn; import org.apache.jena.dboe.trans.data.TransBinaryDataFile; -import org.apache.jena.dboe.transaction.ThreadTxn; import org.apache.jena.dboe.transaction.Transactional; import org.apache.jena.dboe.transaction.TransactionalFactory; import org.apache.jena.dboe.transaction.txn.ComponentId; import org.apache.jena.dboe.transaction.txn.journal.Journal; +import org.apache.jena.system.ThreadTxn; import org.junit.After ; import org.junit.Before ; import org.junit.Test ; - public class TestTransBinaryDataFile { private Journal journal ; private BinaryDataFile baseBinData ; @@ -210,7 +209,7 @@ public class TestTransBinaryDataFile { @Test public void transObjectFile_7() { ThreadTxn.threadTxnWriteAbort(transactional, ()->{ long x1 = transBinData.write(bytes1) ; - }) ; + }).run() ; Txn.executeRead(transactional, ()->{ assertEquals(0L, transBinData.length()) ; http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBlob.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBlob.java b/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBlob.java index e22929c..95be5dd 100644 --- a/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBlob.java +++ b/jena-db/jena-dboe-trans-data/src/test/java/org/apache/jena/dboe/trans/data/TestTransBlob.java @@ -25,12 +25,13 @@ import org.apache.jena.dboe.base.file.BufferChannelMem; import org.apache.jena.dboe.base.file.Location; import org.apache.jena.dboe.jenax.Txn; import org.apache.jena.dboe.trans.data.TransBlob; -import org.apache.jena.dboe.transaction.ThreadTxn; import org.apache.jena.dboe.transaction.Transactional; import org.apache.jena.dboe.transaction.TransactionalFactory; import org.apache.jena.dboe.transaction.txn.ComponentId; import org.apache.jena.dboe.transaction.txn.journal.Journal; import org.apache.jena.query.ReadWrite ; +import org.apache.jena.system.ThreadAction; +import org.apache.jena.system.ThreadTxn; import org.junit.After ; import org.junit.Assert ; import org.junit.Before ; @@ -112,7 +113,7 @@ public class TestTransBlob extends Assert { assertEquals(str1, s2) ; // Start now. - ThreadTxn tt = ThreadTxn.threadTxnRead(transactional, ()-> { + ThreadAction tt = ThreadTxn.threadTxnRead(transactional, ()-> { String sr = transBlob.getString() ; Assert.assertEquals(str1, sr) ; }) ; http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/ThreadTxn.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/ThreadTxn.java b/jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/ThreadTxn.java deleted file mode 100644 index a8a92df..0000000 --- a/jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/ThreadTxn.java +++ /dev/null @@ -1,144 +0,0 @@ -/* - * 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.jena.dboe.transaction; - -import java.util.Objects ; -import java.util.concurrent.Executor ; -import java.util.concurrent.Semaphore ; -import java.util.concurrent.atomic.AtomicReference ; - -import org.apache.jena.dboe.sys.Sys; -import org.apache.jena.query.ReadWrite ; -import org.apache.jena.sparql.core.Transactional ; - -/** - * An action that will happen on a different thread later when {@link #run} is - * called. A thread is created and the transaction started during a call to the - * creation operations {@link #threadTxnRead} or {@link #threadTxnWrite}. - * The associated Runnable is called and the transaction completed when - * {@link #run} is called. Being on a thread, the state of the world the - * forked transaction sees is outside the creating thread which may itself be in a - * transaction. Warning: creating a write transaction inside a write transaction - * will cause deadlock. - */ -public class ThreadTxn { - - /** Create a thread-backed delayed READ transaction action. */ - public static ThreadTxn threadTxnRead(Transactional trans, Runnable action) { - return ThreadTxn.create(trans, ReadWrite.READ, action, false) ; - } - - /** Create a thread-backed delayed WRITE action. - * If called from inside a write transaction on the {@code trans}, - * this will deadlock. - */ - public static ThreadTxn threadTxnWrite(Transactional trans, Runnable action) { - return ThreadTxn.create(trans, ReadWrite.WRITE, action, true) ; - } - - /** Create a thread-backed delayed WRITE-abort action (testing). */ - public static ThreadTxn threadTxnWriteAbort(Transactional trans, Runnable action) { - return ThreadTxn.create(trans, ReadWrite.WRITE, action, false) ; - } - - private final Semaphore semaStart ; - private final Semaphore semaFinish ; - private final AtomicReference<RuntimeException> thrownRuntimeException = new AtomicReference<>(null) ; - private final AtomicReference<Error> thrownError = new AtomicReference<>(null) ; - private final Runnable action ; - - private ThreadTxn(Runnable action) { - this.action = action ; - this.semaStart = new Semaphore(0, true) ; - this.semaFinish = new Semaphore(0, true) ; - } - - /** - * Perform the Runnable, reporting any - * {@link java.lang.RuntimeException} or {@link java.lang.Error} - */ - public void run() { - semaStart.release(); - semaFinish.acquireUninterruptibly() ; - if ( thrownError.get() != null ) - throw thrownError.get() ; - if ( thrownRuntimeException.get() != null ) - throw thrownRuntimeException.get() ; - } - - // Called on the async thread. - private void trigger() { - try { action.run(); } - catch (Error error) { thrownError.set(error) ; throw error ;} - catch (RuntimeException ex) { thrownRuntimeException.set(ex) ; throw ex ; } - } - - // System-shared executor. - private static Executor executor = Sys.executor ; - - /*package*/ static ThreadTxn create(Transactional trans, ReadWrite mode, Runnable action, boolean isCommit) { - Objects.requireNonNull(trans) ; - Objects.requireNonNull(mode) ; - Objects.requireNonNull(action) ; - - ThreadTxn threadAction = new ThreadTxn(action) ; - // Startup semaphore so that the thread has started by the - // time we exit this setup function. - Semaphore semaStartup = new Semaphore(0, true) ; - executor.execute( ()-> { - // NB. trans.begin then semaStartup.release() ; - // This ensures that the transaction has really started. - trans.begin(mode) ; - - // Signal the creator (see below) that the transaction has started. - semaStartup.release() ; - - // Wait for the signal to run the action. - threadAction.semaStart.acquireUninterruptibly(); - - try { - // Performane the action, catch and record any RuntimeException or Error. - threadAction.trigger() ; - - // Finish transaction (if no throwable) - if ( mode == ReadWrite.WRITE ) { - if ( isCommit ) - trans.commit(); - else - trans.abort() ; - trans.end() ; - } else { - // Read - if ( isCommit ) - trans.commit(); - trans.end() ; - } - } - catch (Throwable ex) { - // Suppress now it has trigger transaction mechanism in - // the presence of an unchecked exception. - // Passed to the main thread via ThreadTxn - } - finally { threadAction.semaFinish.release() ; } - }) ; - // Don't return until the transaction has started. - semaStartup.acquireUninterruptibly(); - return threadAction ; - } -} http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestThreadingTransactions.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestThreadingTransactions.java b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestThreadingTransactions.java index de464c0..8b2bf64 100644 --- a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestThreadingTransactions.java +++ b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestThreadingTransactions.java @@ -22,7 +22,8 @@ import java.util.concurrent.Semaphore ; import org.apache.jena.dboe.base.file.Location; import org.apache.jena.dboe.jenax.Txn; -import org.apache.jena.dboe.transaction.ThreadTxn; +import org.apache.jena.system.ThreadAction; +import org.apache.jena.system.ThreadTxn; import org.apache.jena.dboe.transaction.txn.TransactionCoordinator; import org.apache.jena.query.ReadWrite ; import org.junit.After ; @@ -56,7 +57,7 @@ public class TestThreadingTransactions { Assert.assertEquals(label, expected, x); } - ThreadTxn threadRead(String label, TransactionalInteger trans, long expectedValue) { + ThreadAction threadRead(String label, TransactionalInteger trans, long expectedValue) { return ThreadTxn.threadTxnRead(trans, ()->{ read(label, trans, expectedValue) ; }) ; @@ -77,8 +78,8 @@ public class TestThreadingTransactions { @Test public void threadTrans_03() { Semaphore semaBefore = new Semaphore(0, true) ; Semaphore semaAfter = new Semaphore(0, true) ; - ThreadTxn async1 = threadRead("[03/1]", transInt, InitValue); - ThreadTxn async2 = threadRead("[03/2]", transInt, InitValue); + ThreadAction async1 = threadRead("[03/1]", transInt, InitValue); + ThreadAction async2 = threadRead("[03/2]", transInt, InitValue); transInt.begin(ReadWrite.WRITE) ; read("[03/3]", transInt, InitValue) ; @@ -100,13 +101,13 @@ public class TestThreadingTransactions { Semaphore semaBefore2 = new Semaphore(0, true) ; Semaphore semaAfter = new Semaphore(0, true) ; - ThreadTxn async1 = threadRead("[04/1]", transInt, InitValue); - ThreadTxn async2 = threadRead("[04/2]", transInt, InitValue); - ThreadTxn async3 = threadRead("[04/3]", transInt, InitValue); + ThreadAction async1 = threadRead("[04/1]", transInt, InitValue); + ThreadAction async2 = threadRead("[04/2]", transInt, InitValue); + ThreadAction async3 = threadRead("[04/3]", transInt, InitValue); Txn.executeWrite(transInt, transInt::inc); - ThreadTxn async4 = threadRead("[04/3]", transInt, InitValue+1); + ThreadAction async4 = threadRead("[04/3]", transInt, InitValue+1); async1.run() ; Txn.executeWrite(transInt, transInt::inc); // ++ http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTransactionCoordinatorControl.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTransactionCoordinatorControl.java b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTransactionCoordinatorControl.java index c8915a8..977a45c 100644 --- a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTransactionCoordinatorControl.java +++ b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTransactionCoordinatorControl.java @@ -26,11 +26,11 @@ import static org.junit.Assert.assertTrue ; import java.util.concurrent.Semaphore ; import java.util.concurrent.atomic.AtomicInteger ; - +import org.apache.jena.system.ThreadTxn; +import org.apache.jena.system.ThreadAction; import org.apache.jena.dboe.base.file.Location; import org.apache.jena.dboe.jenax.Txn; import org.apache.jena.dboe.migrate.L; -import org.apache.jena.dboe.transaction.ThreadTxn; import org.apache.jena.dboe.transaction.Transactional; import org.apache.jena.dboe.transaction.txn.Transaction; import org.apache.jena.dboe.transaction.txn.TransactionCoordinator; @@ -65,7 +65,7 @@ public class TestTransactionCoordinatorControl { AtomicInteger counter2 = new AtomicInteger(0) ; txnMgr.blockWriters(); - ThreadTxn threadTxn1 = ThreadTxn.threadTxnRead(unit, ()->counter1.incrementAndGet()) ; + ThreadAction threadTxn1 = ThreadTxn.threadTxnRead(unit, ()->counter1.incrementAndGet()) ; threadTxn1.run() ; assertEquals(1, counter1.get()) ; } @@ -126,7 +126,7 @@ public class TestTransactionCoordinatorControl { @Test public void txn_coord_exclusive_2() { AtomicInteger counter1 = new AtomicInteger(0) ; Semaphore finalSema = new Semaphore(0) ; - ThreadTxn ttxn = ThreadTxn.threadTxnWrite(unit, ()->{ + ThreadAction ttxn = ThreadTxn.threadTxnWrite(unit, ()->{ counter1.incrementAndGet() ; }) ; boolean b = txnMgr.tryExclusiveMode(false); http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnLib.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnLib.java b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnLib.java index 9a6dc92..e8bf804 100644 --- a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnLib.java +++ b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnLib.java @@ -22,8 +22,9 @@ import static org.junit.Assert.assertEquals ; import static org.junit.Assert.assertNotEquals ; import static org.junit.Assert.fail ; +import org.apache.jena.system.ThreadTxn; +import org.apache.jena.system.ThreadAction; import org.apache.jena.dboe.jenax.Txn; -import org.apache.jena.dboe.transaction.ThreadTxn; import org.apache.jena.query.ReadWrite ; import org.junit.Test ; @@ -145,23 +146,23 @@ public class TestTxnLib extends AbstractTestTxn { // Tests for thread transactions. @Test public void libTxnThread_1() { - ThreadTxn t = ThreadTxn.threadTxnRead(unit, ()->{}) ; + ThreadAction t = ThreadTxn.threadTxnRead(unit, ()->{}) ; t.run(); } @Test public void libTxnThread_2() { - ThreadTxn t = ThreadTxn.threadTxnWrite(unit, ()-> fail("")) ; + ThreadAction t = ThreadTxn.threadTxnWrite(unit, ()-> fail("")) ; } @Test(expected=AssertionError.class) public void libTxnThread_3() { - ThreadTxn t = ThreadTxn.threadTxnWrite(unit, ()-> fail("")) ; + ThreadAction t = ThreadTxn.threadTxnWrite(unit, ()-> fail("")) ; t.run() ; } @Test public void libTxnThread_10() { long x1 = counter1.get() ; - ThreadTxn t = ThreadTxn.threadTxnWrite(unit, ()->{ counter1.inc() ;}) ; + ThreadAction t = ThreadTxn.threadTxnWrite(unit, ()->{ counter1.inc() ;}) ; long x2 = counter1.get() ; assertEquals("x2", x1, x2) ; t.run() ; @@ -174,7 +175,7 @@ public class TestTxnLib extends AbstractTestTxn { Txn.executeWrite(unit, ()->{ counter1.inc(); // Read the "before" state - ThreadTxn t = ThreadTxn.threadTxnRead(unit, ()->{ long z1 = counter1.get() ; assertEquals("Thread read", x1, z1) ; }) ; + ThreadAction t = ThreadTxn.threadTxnRead(unit, ()->{ long z1 = counter1.get() ; assertEquals("Thread read", x1, z1) ; }) ; counter1.inc(); t.run(); }) ; @@ -184,7 +185,7 @@ public class TestTxnLib extends AbstractTestTxn { @Test public void libTxnThread_12() { long x1 = counter1.get() ; - ThreadTxn t = ThreadTxn.threadTxnRead(unit, () -> { + ThreadAction t = ThreadTxn.threadTxnRead(unit, () -> { long z1 = counter1.get() ; assertEquals("Thread", x1, z1) ; }) ; http://git-wip-us.apache.org/repos/asf/jena/blob/74808450/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnSwitching.java ---------------------------------------------------------------------- diff --git a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnSwitching.java b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnSwitching.java index b68fa7c..96f1e9c 100644 --- a/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnSwitching.java +++ b/jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnSwitching.java @@ -23,7 +23,8 @@ import static org.junit.Assert.fail ; import org.apache.jena.dboe.base.file.Location; import org.apache.jena.dboe.jenax.Txn; -import org.apache.jena.dboe.transaction.ThreadTxn; +import org.apache.jena.system.ThreadTxn; +import org.apache.jena.system.ThreadAction; import org.apache.jena.dboe.transaction.TransInteger; import org.apache.jena.dboe.transaction.txn.*; import org.apache.jena.dboe.transaction.txn.journal.Journal; @@ -97,7 +98,7 @@ public class TestTxnSwitching { integer.inc(); assertEquals(z+3, integer.get()) ; - ThreadTxn threadTxn = ThreadTxn.threadTxnRead(transactional, ()->assertEquals(z+1, integer.get())) ; + ThreadAction threadTxn = ThreadTxn.threadTxnRead(transactional, ()->assertEquals(z+1, integer.get())) ; threadTxn.run() ; transactional.commit() ; @@ -140,8 +141,8 @@ public class TestTxnSwitching { transactional.begin(ReadWrite.READ); TransactionCoordinatorState txnStateR1 = transactional.detach() ; - ThreadTxn t1 = ThreadTxn.threadTxnRead(transactional, ()->assertEquals(z, integer.get() )) ; - ThreadTxn t2 = ThreadTxn.threadTxnRead(transactional, ()->assertEquals(z, integer.get() )) ; + ThreadAction t1 = ThreadTxn.threadTxnRead(transactional, ()->assertEquals(z, integer.get() )) ; + ThreadAction t2 = ThreadTxn.threadTxnRead(transactional, ()->assertEquals(z, integer.get() )) ; transactional.begin(ReadWrite.WRITE); integer.inc();
