Merge pull request #602 from afs/jena1746-tdb2-abort

JENA-1746: TDB2 abort

jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/TransactionalFactory.java

@@ -42,7 +42,7 @@ public static Transactional createTransactional(Journal journal, TransactionalCo
         return createTransactional(coord, elements);
     }
 
-    private static Transactional createTransactional(TransactionCoordinator coord, TransactionalComponent[] elements) {
+    private static Transactional createTransactional(TransactionCoordinator coord, TransactionalComponent... elements) {
         for ( TransactionalComponent tc : elements ) {
             coord.add(tc);
         }

jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/txn/ComponentGroup.java

@@ -18,10 +18,7 @@
 
 package org.apache.jena.dboe.transaction.txn;
 
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
@@ -39,6 +36,10 @@ private ComponentGroup(Map<ComponentId, TransactionalComponent> group) {
         this.group.putAll(group);
     }
 
+    public void addAll(Collection<TransactionalComponent> components) {
+        components.forEach(this::add);
+    }
+
     public void add(TransactionalComponent component) {
         Objects.requireNonNull(component);
         //Log.info(this, "add("+component.getComponentId()+")");

jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/txn/Transaction.java

@@ -121,7 +121,7 @@ public boolean promote(boolean readCommitted) {
         checkState(ACTIVE);
         if ( txnType == TxnType.READ )
             return false;
-        boolean b = txnMgr.promoteTxn(this, readCommitted);
+        boolean b = txnMgr.executePromote(this, readCommitted);
         if ( !b )
             return false;
         mode = ReadWrite.WRITE;
@@ -160,6 +160,7 @@ public void commit() {
             prepare();
         checkState(PREPARE);
         setState(COMMIT);
+        // Sys abort -> state?
         switch(mode) {
             case WRITE:
                 txnMgr.executeCommit(this,
@@ -195,10 +196,12 @@ public void abort() {
     }
 
     public void end() {
+        // [1746]
+        // txnMgr.executeEnd(thus, ()->{});
         txnMgr.notifyEndStart(this);
         if ( isWriteTxn() && getState() == ACTIVE ) {
             //Log.warn(this, "Write transaction with no commit() or abort() before end()");
-            // Just the abort process.
+            // Just abort process.
             abort$();
             endInternal();
             throw new TransactionException("Write transaction with no commit() or abort() before end() - forced abort");

jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/txn/TransactionCoordinator.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Consumer;
 
 import org.apache.jena.atlas.logging.FmtLog;
 import org.apache.jena.atlas.logging.Log;
@@ -75,8 +76,9 @@ public class TransactionCoordinator {
     private boolean configurable = true;
 
     private final ComponentGroup components = new ComponentGroup();
+    private final List<TransactionListener> listeners = new ArrayList<>();
+
     // Components
-    private ComponentGroup txnComponents = null;
     private List<ShutdownHook> shutdownHooks;
     private TxnIdGenerator txnIdGenerator = TxnIdFactory.txnIdGenSimple;
 
@@ -138,10 +140,8 @@ public TransactionCoordinator(Journal journal, List<TransactionalComponent> comp
     private TransactionCoordinator(Journal journal, List<TransactionalComponent> txnComp, List<ShutdownHook> shutdownHooks) {
         this.journal = journal;
         this.shutdownHooks = new ArrayList<>(shutdownHooks);
-        if ( txnComp != null ) {
-            //txnComp.forEach(x-> System.out.println(x.getComponentId().label()+" :: "+Bytes.asHex(x.getComponentId().bytes()) ) );
-            txnComp.forEach(components::add);
-        }
+        if ( txnComp != null )
+            components.addAll(txnComp);
     }
 
     /** Add a {@link TransactionalComponent}.
@@ -167,6 +167,18 @@ public TransactionCoordinator remove(TransactionalComponent elt) {
         return this;
     }
 
+    public TransactionCoordinator addListener(TransactionListener listener) {
+        checklAllowModification();
+        listeners.add(listener);
+        return this;
+    }
+
+    public TransactionCoordinator removeListener(TransactionListener listener) {
+        checklAllowModification();
+        listeners.remove(listener);
+        return this;
+    }
+
     /**
      * Perform modification of this {@code TransactionCoordiator} after it has been
      * started.
@@ -179,7 +191,7 @@ public TransactionCoordinator remove(TransactionalComponent elt) {
      * <p>
      * Use with care!
      */
-    public void modify(Runnable action) {
+    public void modifyConfig(Runnable action) {
         try {
             startExclusiveMode();
             configurable = true;
@@ -190,6 +202,11 @@ public void modify(Runnable action) {
         }
     }
 
+    /** Call the action for each listener */ 
+    private void listeners(Consumer<TransactionListener> action) {
+        listeners.forEach(x->action.accept(x));
+    }
+
     /**
      * Add a shutdown hook. Shutdown is not guaranteed to be called
      * and hence hooks may not get called.
@@ -533,6 +550,7 @@ public Transaction begin(TxnType txnType, boolean canBlock) {
         Transaction transaction = begin$(txnType);
         startActiveTransaction(transaction);
         transaction.begin();
+        notifyBegin(transaction);
         return transaction;
     }
 
@@ -589,17 +607,21 @@ private ComponentGroup chooseComponents(ComponentGroup components, TxnType txnTy
     }
 
     /** Attempt to promote a transaction from READ mode to WRITE mode based.
-     *  Whether intevening commits are seen is determined by the boolean flag.
-     * Return true if the transaction is already a writer.
+     *  Whether intervening commits are seen is determined by the boolean flag.
+     *  Return true if the transaction is already a writer.
      */
-    /*package*/ boolean promoteTxn(Transaction transaction, boolean readCommittedPromotion) {
+    /*package*/ boolean executePromote(Transaction transaction, boolean readCommittedPromotion) {
         if ( transaction.getMode() == ReadWrite.WRITE )
             return true;
-        // While this code allows promotion of TxnType.READ, this ability is usually rejected
+        // Even if promotion of TxnType.READ allowed, this ability is usually rejected
         // by the transaction system around it. e.g. TransactionalBase.
         if ( transaction.getTxnType() == TxnType.READ )
             throw new TransactionException("promote: can't promote a READ transaction");
-        return promoteTxn$(transaction, readCommittedPromotion);
+
+        notifyPromoteStart(transaction);
+        boolean b = promoteTxn$(transaction, readCommittedPromotion);
+        notifyPromoteFinish(transaction);
+        return b;
     }
 
     private boolean promoteTxn$(Transaction transaction, boolean readCommittedPromotion) {
@@ -689,8 +711,11 @@ private boolean promotionWaitForWriters() {
     /*package*/ void completed(Transaction transaction) {
         finishActiveTransaction(transaction);
         journal.reset();
+        notifyEnd(transaction);
     }
 
+    // Internally, an APi call "commit" is "prepare then commit". 
+
     /*package*/ void executePrepare(Transaction transaction) {
         // Do here because it needs access to the journal.
         notifyPrepareStart(transaction);
@@ -705,7 +730,13 @@ private boolean promotionWaitForWriters() {
     }
 
     /*package*/ void executeCommit(Transaction transaction, Runnable commit, Runnable finish, Runnable sysabort) {
+        notifyCommitStart(transaction);
         if ( transaction.getMode() == ReadWrite.READ ) {
+
+            //[1746]
+            //executeCommitReader();
+            // No commit on components, all "end".
+            // Make abort the same?
             finish.run();
             notifyCommitFinish(transaction);
             return;
@@ -719,6 +750,7 @@ private boolean promotionWaitForWriters() {
         } catch (Throwable th) {
             throw th;
         } finally { journal.endWrite(); }
+        notifyCommitFinish(transaction);
     }
 
     private void executeCommitWriter(Transaction transaction, Runnable commit, Runnable finish, Runnable sysabort) {
@@ -777,7 +809,6 @@ private void executeCommitWriter(Transaction transaction, Runnable commit, Runna
             finish.run();
             // Bump global serialization point
             advanceDataVersion();
-            notifyCommitFinish(transaction);
         }
     }
 
@@ -879,38 +910,70 @@ private void finishActiveTransaction(Transaction transaction) {
     public long countActive()           { return activeTransactionCount.get(); }
 
     // notify*Start/Finish called round each transaction lifecycle step
-    // Called in cooperation between Transaction and TransactionCoordinator
-    // depending on who is actually do the work of each step.
 
-    /*package*/ void notifyPrepareStart(Transaction transaction) {}
+    private void notifyBegin(Transaction transaction) {
+        listeners(x -> x.notifyTxnStart(transaction));
+    }
 
-    /*package*/ void notifyPrepareFinish(Transaction transaction) {}
+    private void notifyEnd(Transaction transaction) {
+        listeners(x -> x.notifyTxnFinish(transaction));
+    }
+
+    private void notifyPromoteStart(Transaction transaction) {
+        listeners(x -> x.notifyPromoteStart(transaction));
+    }
+
+    private void notifyPromoteFinish(Transaction transaction) {
+        listeners(x -> x.notifyPromoteFinish(transaction));
+    }
+
+    private void notifyPrepareStart(Transaction transaction) {
+        listeners(x -> x.notifyPrepareStart(transaction));
+    }
+
+    private void notifyPrepareFinish(Transaction transaction) {
+        listeners(x -> x.notifyPrepareFinish(transaction));
+    }
 
     // Writers released here - can happen because of commit() or abort().
 
-    private void notifyCommitStart(Transaction transaction) {}
+    private void notifyCommitStart(Transaction transaction) {
+        listeners(x -> x.notifyCommitStart(transaction));
+    }
 
-    private void notifyCommitFinish(Transaction transaction) {
+    private void notifyCommitFinish(Transaction transaction)    {
+        listeners(x->x.notifyCommitFinish(transaction));
         if ( transaction.getMode() == ReadWrite.WRITE )
             releaseWriterLock();
     }
 
-    private void notifyAbortStart(Transaction transaction) { }
+    private void notifyAbortStart(Transaction transaction) { 
+        listeners(x->x.notifyAbortStart(transaction));
+    }
 
     private void notifyAbortFinish(Transaction transaction) {
+        listeners(x->x.notifyAbortFinish(transaction));
         if ( transaction.getMode() == ReadWrite.WRITE )
             releaseWriterLock();
     }
 
-    /*package*/ void notifyEndStart(Transaction transaction) { }
+    /*package*/ void notifyEndStart(Transaction transaction) {
+        listeners(x->x.notifyEndStart(transaction));
+    }
 
-    /*package*/ void notifyEndFinish(Transaction transaction) {}
+    /*package*/ void notifyEndFinish(Transaction transaction) {
+        listeners(x->x.notifyEndFinish(transaction));
+    }
 
     // Called by Transaction once at the end of first commit()/abort() or end()
 
-    /*package*/ void notifyCompleteStart(Transaction transaction) { }
+    /*package*/ void notifyCompleteStart(Transaction transaction) {
+        listeners(x -> x.notifyCompleteStart(transaction));
+    }
 
-    /*package*/ void notifyCompleteFinish(Transaction transaction) { }
+    /*package*/ void notifyCompleteFinish(Transaction transaction) {
+        listeners(x -> x.notifyCompleteFinish(transaction));
+    }
 
     // Coordinator state.
     private final AtomicLong countBegin         = new AtomicLong(0);

jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/txn/TransactionListener.java

@@ -0,0 +1,59 @@
+/*
+ * 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.txn;
+
+/** Call backs for the transaction lifecycle. */ 
+public interface TransactionListener {
+
+    /** A transaction has started; begin has done the setup. */
+    public default void notifyTxnStart(Transaction transaction) { }
+
+    /** Start a call to promote */
+    public default void notifyPromoteStart(Transaction transaction) { }
+    /** Finish a call to promote */
+    public default void notifyPromoteFinish(Transaction transaction) { }
+
+    /** Start prepare during a commit */
+    public default void notifyPrepareStart(Transaction transaction) { }
+    /** Finish prepare during a commit */
+    public default void notifyPrepareFinish(Transaction transaction) { }
+
+    /** Start a commit (prepare has been done) */
+    public default void notifyCommitStart(Transaction transaction) { }
+    /** Finish a commit (prepare has been done) */
+    public default void notifyCommitFinish(Transaction transaction) { }
+
+    /** Start an abort */
+    public default void notifyAbortStart(Transaction transaction) { }
+    /** Start finish an abort */
+    public default void notifyAbortFinish(Transaction transaction) { }
+
+    /** Start an end() */
+    public default void notifyEndStart(Transaction transaction) { }
+    /** Finish an end() */
+    public default void notifyEndFinish(Transaction transaction) { }
+
+    /** Start the complete step. */
+    public default void notifyCompleteStart(Transaction transaction) { }
+    /** Finish the complete step. */
+    public default void notifyCompleteFinish(Transaction transaction) { }
+
+    /** Transaction has finished. This is called during "complete" */
+    public default void notifyTxnFinish(Transaction transaction) { }
+}

jena-db/jena-dboe-transaction/src/main/java/org/apache/jena/dboe/transaction/txn/TransactionalComponentBase.java

@@ -50,6 +50,9 @@ protected X _begin(ReadWrite readWrite, TxnId txnId) {
         return null;
     }
 
+    @Override
+    protected X _promote(TxnId txnId, X state) { return null; }
+
     @Override
     protected ByteBuffer _commitPrepare(TxnId txnId, X state) {
         return null;
@@ -69,9 +72,5 @@ protected void _complete(TxnId txnId, X state) {}
 
     @Override
     protected void _shutdown() {}
-
-    @Override
-    protected X _promote(TxnId txnId, X state) { return null; }
-
 }
 

jena-db/jena-dboe-transaction/src/test/java/org/apache/jena/dboe/transaction/TestTxnSwitching.java

@@ -40,7 +40,7 @@ public class TestTxnSwitching {
 
     //Transactional transactional = TransactionalFactory.create(jrnl, integer);
     TransactionalBase transactional;
-    TransactionCoordinator txnMgr = new  TransactionCoordinator(jrnl); {
+    TransactionCoordinator txnMgr = new TransactionCoordinator(jrnl); {
         txnMgr.add(integer);
         transactional = new TransactionalBase(txnMgr);
         txnMgr.start();