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 2017/09/28 16:05:39 UTC
[027/100] [abbrv] [partial] jena git commit: JENA-1397: Rename
modules and set versions
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransMonitor.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransMonitor.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransMonitor.java
new file mode 100644
index 0000000..8c44724
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransMonitor.java
@@ -0,0 +1,192 @@
+/*
+ * 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.seaborne.dboe.transaction;
+
+import java.io.PrintStream ;
+import java.nio.ByteBuffer ;
+import java.util.LinkedHashMap ;
+import java.util.Map ;
+import java.util.concurrent.atomic.LongAdder ;
+import java.util.stream.Collectors ;
+
+import org.seaborne.dboe.transaction.txn.ComponentId ;
+import org.seaborne.dboe.transaction.txn.SysTransState ;
+import org.seaborne.dboe.transaction.txn.Transaction ;
+import org.seaborne.dboe.transaction.txn.TransactionalComponent ;
+
+/** This class is stateless in the transaction but it records counts of
+ * every {@link TransactionalComponent} operations.
+ * For each operation called "ABC" there is a member field "counterABC".
+ */
+public class TransMonitor implements TransactionalComponent {
+
+ // LongAdder for stats (better from mostly write, not consistent read)
+
+ /** Counters, in initialization order */
+ private Map<String, LongAdder> counters = new LinkedHashMap<>() ;
+ private final ComponentId componentId ;
+
+ private LongAdder allocCounter(String string) {
+ LongAdder counter = new LongAdder() ;
+ counters.put(string, counter) ;
+ return counter ;
+ }
+ public TransMonitor(ComponentId cid) {
+ this.componentId = cid ;
+ }
+
+ /** Reset all counters to zero */
+ public void reset() {
+ counters.forEach( (s,c) -> c.reset()) ;
+ }
+
+ /** Get a copy of the counters with current values.
+ * The values are as of the point of this being called and are not changed
+ * by any later calls to monitored operations.
+ */
+ public Map<String, Long> getAll() {
+ return counters.entrySet().stream()
+ .collect(Collectors.toMap(
+ e -> e.getKey(),
+ e -> e.getValue().sum()
+ ));
+ }
+
+ /** Print the counters state. */
+ public void print() {
+ print(System.out) ;
+ }
+
+ /** Print the counters state. */
+ public void print(PrintStream ps) {
+ ps.println("Transaction Counters:") ;
+ counters.forEach( (s,c) -> {
+ ps.printf(" %-15s %4d\n", s, c.longValue()) ;
+ }) ;
+ }
+
+ public LongAdder counterGetComponentId = allocCounter("getComponentId") ;
+
+ @Override
+ public ComponentId getComponentId() {
+ counterGetComponentId.increment() ;
+ return componentId ;
+ }
+
+ public LongAdder counterStartRecovery = allocCounter("startRecovery") ;
+
+ @Override
+ public void startRecovery() {
+ counterStartRecovery.increment() ;
+ }
+
+ public LongAdder counterRecover = allocCounter("recover") ;
+
+ @Override
+ public void recover(ByteBuffer ref) {
+ counterRecover.increment() ;
+ }
+
+ public LongAdder counterFinishRecovery = allocCounter("finishRecovery") ;
+
+ @Override
+ public void finishRecovery() {
+ counterFinishRecovery.increment() ;
+ }
+
+ public LongAdder counterCleanStart = allocCounter("finishRecovery") ;
+
+ @Override
+ public void cleanStart() {
+ counterCleanStart.increment() ;
+ }
+
+ public LongAdder counterBegin = allocCounter("begin") ;
+
+ @Override
+ public void begin(Transaction transaction) {
+ counterBegin.increment() ;
+ }
+
+ public LongAdder counterPromote = allocCounter("promote") ;
+
+ @Override
+ public boolean promote(Transaction transaction) {
+ counterPromote.increment() ;
+ return true ;
+ }
+
+ public LongAdder counterCommitPrepare = allocCounter("commitPrepare") ;
+
+ @Override
+ public ByteBuffer commitPrepare(Transaction transaction) {
+ counterCommitPrepare.increment() ;
+ return null ;
+ }
+
+ public LongAdder counterCommit = allocCounter("commit") ;
+
+ @Override
+ public void commit(Transaction transaction) {
+ counterCommit.increment() ;
+ }
+
+ public LongAdder counterCommitEnd = allocCounter("commitEnd") ;
+
+ @Override
+ public void commitEnd(Transaction transaction) {
+ counterCommitEnd.increment() ;
+ }
+
+ public LongAdder counterAbort = allocCounter("abort") ;
+
+ @Override
+ public void abort(Transaction transaction) {
+ counterAbort.increment() ;
+ }
+
+ public LongAdder counterComplete = allocCounter("complete") ;
+
+ @Override
+ public void complete(Transaction transaction) {
+ counterComplete.increment() ;
+ }
+
+ public LongAdder counterDetach = allocCounter("detach") ;
+
+ @Override
+ public SysTransState detach() {
+ counterDetach.increment() ;
+ return null ;
+ }
+
+ public LongAdder counterAttach = allocCounter("attach") ;
+
+ @Override
+ public void attach(SysTransState systemState) {
+ counterAttach.increment() ;
+ }
+
+ public LongAdder counterShutdown = allocCounter("shutdown") ;
+
+ @Override
+ public void shutdown() {
+ counterShutdown.increment() ;
+ }
+}
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/Transactional.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/Transactional.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/Transactional.java
new file mode 100644
index 0000000..8242e3c
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/Transactional.java
@@ -0,0 +1,99 @@
+/*
+ * 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.seaborne.dboe.transaction;
+
+import org.apache.jena.query.ReadWrite ;
+import org.seaborne.dboe.jenax.Txn ;
+
+/** Interface that encapsulates the begin/abort|commit/end operations.
+ * <p>The read lifecycle is:
+ * <pre> begin(READ) ... end()</pre>
+ * <p>{@code commit} and {@code abort} are allowed.
+ * <p>The write lifecycle is:
+ * <pre> begin(WRITE) ... abort() or commit()</pre>
+ * <p>{@code end()} is optional but preferred.
+ * <p>
+ * Helper code is available {@link Txn} so, for example:
+ * <pre>Txn.execRead(dataset, ()-> { ... sparql query ... });</pre>
+ * <pre>Txn.execWrite(dataset, ()-> { ... sparql update ... });</pre>
+ * <p>
+ * Directly called, code might look like:
+ * <pre>
+ * Transactional object = ...
+ * object.begin(ReadWrite.READ) ;
+ * try {
+ * ... actions inside a read transaction ...
+ * } finally { object.end() ; }
+ * </pre>
+ * or
+ * <pre>
+ * Transactional object = ...
+ * object.begin(ReadWrite.WRITE) ;
+ * try {
+ * ... actions inside a write transaction ...
+ * object.commit() ;
+ * } finally {
+ * // This causes an abort if {@code commit} has not been called.
+ * object.end() ;
+ * }
+ * </pre>
+ * Exceptions will not be thrown.
+ * If any do occur, this indicates serious internal problems with the transaction system.
+ */
+public interface Transactional extends org.apache.jena.sparql.core.Transactional
+{
+ /** Start either a READ or WRITE transaction */
+ @Override
+ public void begin(ReadWrite readWrite) ;
+
+ /** Attempt to promote a read transaction to a write transaction.
+ * This is not guaranteed to succeed - any changes by another write transaction
+ * may restrict promotion.
+ * <p>
+ * In the MR+SW implementation, any intervening write transaction will block promotion.
+ * <p>
+ * Promoting a transaction which is already a write transaction will return true.
+ * <p>
+ * Consider also:
+ * <pre>
+ * .end() ;
+ * .begin(WRITE) ;
+ * </pre>
+ * to see any intermediate commits from another writer.
+ *
+ * @return boolean indicating whether the transaction is now a write transaction or not.
+ */
+ public boolean promote() ;
+
+ /** Commit a transaction - finish the transaction and make any changes permanent (if a "write" transaction) */
+ @Override
+ public void commit() ;
+
+ /** Abort a transaction - finish the transaction and undo any changes (if a "write" transaction) */
+ @Override
+ public void abort() ;
+
+ /** Finish the transaction - if a write transaction and commit() has not been called, then abort. */
+ @Override
+ public void end() ;
+
+ /** Say whether inside a transaction. */
+ @Override
+ public boolean isInTransaction() ;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalFactory.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalFactory.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalFactory.java
new file mode 100644
index 0000000..4af4167
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalFactory.java
@@ -0,0 +1,60 @@
+/*
+ * 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.seaborne.dboe.transaction;
+
+import org.seaborne.dboe.base.file.Location ;
+import org.seaborne.dboe.transaction.txn.TransactionCoordinator ;
+import org.seaborne.dboe.transaction.txn.TransactionalBase ;
+import org.seaborne.dboe.transaction.txn.TransactionalComponent ;
+import org.seaborne.dboe.transaction.txn.TransactionalSystem ;
+import org.seaborne.dboe.transaction.txn.journal.Journal ;
+
+/** Helper operations for creating a {@link Transactional}.
+ * The operations capture some common patterns.
+ */
+public class TransactionalFactory {
+
+ /** Create, and start, management of a number of {@link TransactionalComponent}s */
+ public static Transactional createTransactional(Location location, TransactionalComponent ... elements) {
+ TransactionCoordinator coord = new TransactionCoordinator(location) ;
+ return createTransactional(coord, elements) ;
+ }
+
+ /** Create, and start, management of a number of {@link TransactionalComponent}s */
+ public static Transactional createTransactional(Journal journal, TransactionalComponent ... elements) {
+ TransactionCoordinator coord = new TransactionCoordinator(journal) ;
+ return createTransactional(coord, elements) ;
+ }
+
+ private static Transactional createTransactional(TransactionCoordinator coord, TransactionalComponent[] elements) {
+ for ( TransactionalComponent tc : elements ) {
+ coord.add(tc) ;
+ }
+ TransactionalBase base = new TransactionalBase(coord) ;
+ coord.start() ;
+ return base ;
+ }
+
+ /** Create, but do not start, a {@link TransactionalSystem} from a {@link TransactionCoordinator} */
+ public static TransactionalSystem createTransactionalSystem(TransactionCoordinator coord) {
+ return new TransactionalBase(coord) ;
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalMonitor.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalMonitor.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalMonitor.java
new file mode 100644
index 0000000..5b641a6
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/TransactionalMonitor.java
@@ -0,0 +1,39 @@
+/*
+ * 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.seaborne.dboe.transaction;
+
+import org.apache.jena.query.ReadWrite ;
+
+/** Interface for the Transactional interface */
+public interface TransactionalMonitor {
+ default void startBegin(ReadWrite mode) {}
+ default void finishBegin(ReadWrite mode) {}
+
+ default void startPromote() {}
+ default void finishPromote() {}
+
+ default void startCommit() {}
+ default void finishCommit() {}
+
+ default void startAbort() {}
+ default void finishAbort() {}
+
+ default void startEnd() {}
+ default void finishEnd() {}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentGroup.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentGroup.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentGroup.java
new file mode 100644
index 0000000..6d78127
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentGroup.java
@@ -0,0 +1,78 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import java.util.Arrays ;
+import java.util.HashMap ;
+import java.util.Map ;
+import java.util.Objects ;
+import java.util.function.BiConsumer ;
+import java.util.function.Consumer ;
+
+import org.apache.jena.atlas.logging.Log ;
+
+final
+public class ComponentGroup {
+ private Map<ComponentId, TransactionalComponent> group = new HashMap<>() ;
+
+ public ComponentGroup(TransactionalComponent...components) {
+ Arrays.asList(components).forEach(this::add);
+ }
+
+ private ComponentGroup(Map<ComponentId, TransactionalComponent> group) {
+ this.group.putAll(group);
+ }
+
+ public void add(TransactionalComponent component) {
+ Objects.requireNonNull(component) ;
+ //Log.info(this, "add("+component.getComponentId()+")") ;
+ if ( component.getComponentId() == null )
+ Log.warn(this, "Null component id - likely to be overwritten: "+component) ;
+
+ if ( group.containsKey(component.getComponentId()) ) {
+ Log.warn(this, component.getComponentId().toString()) ;
+ Log.warn(this, "Add component already in the group: "+component) ;
+ }
+
+ group.put(component.getComponentId(), component) ;
+ }
+
+ /*package*/ void remove(ComponentId componentId) {
+ group.remove(componentId) ;
+ }
+
+ public TransactionalComponent findComponent(ComponentId componentId) {
+ return group.get(componentId) ;
+ }
+
+ public void forEachComponent(Consumer<? super TransactionalComponent> action) {
+ group.values().forEach(action) ;
+ }
+
+ public void forEach(BiConsumer<ComponentId, TransactionalComponent> action) {
+ group.forEach(action);
+ }
+
+ public void addAll(ComponentGroup components) {
+ this.group.putAll(components.group);
+ }
+
+ public int size() { return group.size() ; }
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentId.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentId.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentId.java
new file mode 100644
index 0000000..f6fe5ed
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentId.java
@@ -0,0 +1,124 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import java.util.Arrays ;
+import java.util.UUID ;
+
+import org.apache.jena.atlas.lib.Bytes ;
+
+/** A {@code ComponentId} consists of two parts: a globally unique context
+ * (roughly - the domain where the {@code ComponentId} is valid)
+ * and an id within the context.
+ * Context is often a single coordinator but can be, for example,
+ * a distributed transaction coordinator.
+ */
+
+public class ComponentId {
+ // Fixed size.
+ public static final int SIZE = 4 ;
+ private final UUID coordinatorId ;
+ private final byte[] bytes ;
+ // Just a helper for development. Not persisted in the journal.
+ private final String displayName ;
+
+ /** Create a new ComponentId from the given bytes.
+ * The bytes are <em>not</em> copied.
+ * The caller must not modify them after this call.
+ * The static method {@link #create(String, byte[])}
+ * does the copy.
+ */
+ private ComponentId(String label, UUID coordinatorId, byte[] bytes) {
+ this.coordinatorId = coordinatorId ;
+ if ( label == null )
+ label = "" ;
+ if ( bytes.length > SIZE )
+ throw new IllegalArgumentException("Bytes for ComponentId too long "+bytes.length+" > "+SIZE) ;
+ if ( bytes.length < SIZE )
+ // Make safe.
+ bytes = Arrays.copyOf(bytes, SIZE) ;
+ this.bytes = bytes ;
+ this.displayName = label ;
+ }
+
+ public byte[] getBytes() { return bytes ; }
+
+ public UUID getBaseId() { return coordinatorId ; }
+
+ public String label() { return displayName ; }
+
+ @Override
+ public String toString() { return displayName+"["+Bytes.asHex(bytes)+"]" ; }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31 ;
+ int result = 1 ;
+ result = prime * result + Arrays.hashCode(bytes) ;
+ return result ;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if ( this == obj )
+ return true ;
+ if ( obj == null )
+ return false ;
+ if ( getClass() != obj.getClass() )
+ return false ;
+ ComponentId other = (ComponentId)obj ;
+ if ( !Arrays.equals(bytes, other.bytes) )
+ return false ;
+ return true ;
+ }
+
+ /** Create a ComponentId from the given bytes */
+ public static ComponentId create(UUID coordinatorBase, byte[] bytes) {
+ bytes = Arrays.copyOf(bytes, bytes.length) ;
+ return new ComponentId(null, coordinatorBase, bytes) ;
+ }
+
+ /** Given a base componentId, create a derived (different) one.
+ * This is deterministically done based on baseComponentId and index.
+ * The label is just for display purposes.
+ */
+ public static ComponentId alloc(String label, UUID coordinatorBase, int index) {
+ byte[] bytes = Bytes.intToBytes(index) ;
+ return new ComponentId(label, coordinatorBase, bytes) ;
+ }
+
+// private static ComponentId create(byte[] bytes, String label, int index) {
+// bytes = Arrays.copyOf(bytes, bytes.length) ;
+// int x = Bytes.getInt(bytes, bytes.length-SystemBase.SizeOfInt) ;
+// x = x ^ index ;
+// Bytes.setInt(x, bytes, bytes.length - SystemBase.SizeOfInt) ;
+// ComponentId cid = new ComponentId(label+"-"+index, bytes) ;
+// return cid ;
+// }
+
+ static int counter = 0 ;
+ /** Return a fresh ComponentId (not preserved across JVM runs) */
+ public static ComponentId allocLocal() {
+ counter++ ;
+ UUID uuid = UUID.randomUUID() ;
+ return alloc("Local-"+counter, uuid, counter) ;
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentIds.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentIds.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentIds.java
new file mode 100644
index 0000000..2d586a0
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/ComponentIds.java
@@ -0,0 +1,62 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+
+/** Some fixed, marker, {@link ComponentId}s */
+public class ComponentIds {
+
+ public static final ComponentId idNull = make("Monitor", -9999) ;
+ public static final ComponentId idSystem = make("System", -9988) ;
+
+ private static ComponentId make(String string, int v) {
+ return ComponentId.alloc(string, null, v) ;
+ }
+
+
+//
+// // Linux : "uuid -v 4"
+//
+// /* For single instance components and base ids for incremental registration */
+// public static final ComponentId idTxnMRSW = make("MRSW", "93a58341-ed53-4f0c-bac1-d9969ea38cf3") ;
+// // TODO tie to registration (this predates)
+// public static final ComponentId idTxnCounter = make("Counter", "6b901671-e6db-45c5-9217-7506d21a0000") ;
+// public static final ComponentId idMonitor = make("Monitor", "c4d8a1e6-052b-413a-8d80-c5a6b442e608") ;
+// public static final ComponentId idNull = make("Monitor", "e6e31271-b6dc-452c-b624-d4e099464365") ;
+// public static final ComponentId idSystem = make("System", "95e0f729-ad29-48b2-bd70-e3738663c578") ;
+// public static final ComponentId idBlobBase = make("Blob", "43436b91-87ce-4d6b-827c-c3b9ea6536ba") ;
+// public static final ComponentId idDev = make("Blob", "82a6833a-1475-495a-83ca-10370c7c40cd") ;
+//
+// public static ComponentId make(String label, String uuidStr) {
+// byte[] bytes = L.uuidAsBytes(uuidStr) ;
+// return ComponentId.create(label, bytes) ;
+// }
+
+ /* For later
+ 82a6833a-1475-495a-83ca-10370c7c40cd
+ 1d32231b-aa11-47ed-8893-6b36673fe04c
+ 27c4845e-c05a-410d-8c74-278a23b03bbd
+ f841ba46-a297-487d-b622-dd452c888dab
+ 09efba54-7428-4689-929d-b1719a56c345
+ a9fbdc3c-442d-4086-8f40-b6ef773871b9
+ */
+
+
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/PrepareState.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/PrepareState.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/PrepareState.java
new file mode 100644
index 0000000..c839e8f
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/PrepareState.java
@@ -0,0 +1,41 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import java.nio.ByteBuffer ;
+
+final
+public class PrepareState {
+ private final ComponentId component ;
+ private final ByteBuffer data ;
+
+ public /*For testing*/ PrepareState(ComponentId compoent, ByteBuffer data) {
+ this.component = compoent ;
+ this.data = data ;
+ }
+
+ public ComponentId getComponent() {
+ return component ;
+ }
+
+ public ByteBuffer getData() {
+ return data ;
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/QuorumGenerator.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/QuorumGenerator.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/QuorumGenerator.java
new file mode 100644
index 0000000..e214ebd
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/QuorumGenerator.java
@@ -0,0 +1,28 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import org.apache.jena.query.ReadWrite ;
+
+/** Generate a transaction quorum for a transaction as it begins */
+public interface QuorumGenerator {
+ public ComponentGroup genQuorum(ReadWrite mode) ;
+}
+
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrBase.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrBase.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrBase.java
new file mode 100644
index 0000000..7db11b9
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrBase.java
@@ -0,0 +1,176 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import java.nio.ByteBuffer ;
+
+import org.apache.jena.atlas.lib.Closeable ;
+import org.apache.jena.atlas.lib.Sync ;
+import org.seaborne.dboe.base.file.BufferChannel ;
+import org.slf4j.Logger ;
+import org.slf4j.LoggerFactory ;
+
+/** Helper class for the manage the persistent state of a transactional.
+ * The persistent state is assumed to be a fixed size, or at least
+ * of known maximum size.
+ * May not be suitable for all transactional component implementations.
+ */
+
+public abstract class StateMgrBase implements Sync, Closeable {
+ /* Compare to TransBlob which is a similar idea but where it is a
+ * component in the transaction component group. That can lead to
+ * the wrong control, or at least unclear, of the writing during
+ * the transaction commit cycle.
+ */
+
+ private static Logger log = LoggerFactory.getLogger(StateMgrBase.class) ;
+
+ private final BufferChannel storage ;
+ // One ByteBuffer that is reused where possible.
+ // This is short-term usage
+ // * get disk state-> deserialize
+ // * serialize->set disk state
+ // on a single thread (the transaction writer).
+ private ByteBuffer bb ;
+ // Is the internal state out of sync with the disk state?
+ private boolean dirty = false ;
+
+ protected StateMgrBase(BufferChannel storage, int sizeBytes) {
+ bb = ByteBuffer.allocate(sizeBytes) ;
+ this.storage = storage ;
+ }
+
+ /**
+ * After the default initial state is known, call this, for example, at the
+ * end of the constructor. If no on-disk state is found, a clean copy is written.
+ */
+ protected void init() {
+ if ( ! storage.isEmpty() )
+ readState() ;
+ else
+ writeState() ;
+ }
+
+ /* Serialize the necessary state into a ByteBuffer.
+ * The argument ByteBuffer can be used or a new one returned
+ * if it is the wrong size (e.g. too small). The returned one will become the
+ * recycled ByteBuffer. The returned ByteBuffer should have posn/limit
+ * delimiting the necessary space to write.
+ */
+ protected abstract ByteBuffer serialize(ByteBuffer bytes) ;
+
+ /*
+ * Deserialize the persistent state from the ByteBuffer (delimited by posn/limit).
+ * The byte buffer will be the recycled one from last time.
+ * Most
+ */
+ protected abstract void deserialize(ByteBuffer bytes) ;
+
+ /** Note that the in-memory state is not known to be the same
+ * as the on-disk state.
+ */
+ protected void setDirtyFlag() {
+ dirty = true ;
+ }
+
+ /**
+ * Event call for state writing. Called after successful
+ * writing of the state.
+ */
+ protected abstract void writeStateEvent() ;
+
+ /**
+ * Event call for state reading. Called after successful
+ * deserializing of the state.
+ */
+ protected abstract void readStateEvent() ;
+
+ /** Note that the in-memory state is the same
+ * as the on-disk state, or at least the on-disk state is
+ * acceptable for restart at any time.
+ */
+ protected void clearDirtyFlag() {
+ dirty = false ;
+ }
+
+ /** Low level control - for example, used for cloning setup.
+ * Use with care.
+ */
+ public BufferChannel getBufferChannel() {
+ return storage ;
+ }
+
+ /** Return the serialized state using the internal ByteBuffer
+ * Typically called by "prepare" for the bytes to write to the journal.
+ * Calls {@link #serialize}.
+ * This method does not perform an external I/O.
+ */
+ public ByteBuffer getState() {
+ bb.rewind() ;
+ serialize(bb) ;
+ return bb ;
+ }
+
+ /** Set the in-memory state from a ByteBuffer, for example, from journal recovery.
+ * This method does not perform an external I/O.
+ * Call "writeState" to put the n-memory state as the disk state.
+ */
+ public void setState(ByteBuffer buff) {
+ buff.rewind() ;
+ deserialize(buff) ;
+ dirty = true ;
+ }
+
+ //public BufferChannel getChannel() { return storage ; }
+
+ /** The write process : serialize, write, sync,
+ * After this, the bytes definitely are on disk, not in some OS cache
+ */
+ public void writeState() {
+ bb.rewind() ;
+ ByteBuffer bb1 = serialize(bb) ;
+ if ( bb1 != null )
+ bb = bb1 ;
+ bb.rewind() ;
+ int len = storage.write(bb, 0) ;
+ storage.sync() ;
+ dirty = false ;
+ writeStateEvent() ;
+ }
+
+ /** The read process : get all bytes on disk, deserialize */
+ public void readState() {
+ bb.rewind() ;
+ int len = storage.read(bb, 0) ;
+ bb.rewind() ;
+ deserialize(bb) ;
+ readStateEvent() ;
+ }
+
+ @Override
+ public void sync() {
+ if ( dirty )
+ writeState() ;
+ }
+
+ @Override
+ public void close() {
+ storage.close() ;
+ }
+}
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrData.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrData.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrData.java
new file mode 100644
index 0000000..5a26dba
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrData.java
@@ -0,0 +1,88 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import java.nio.ByteBuffer ;
+import java.util.Arrays ;
+
+import org.seaborne.dboe.base.file.BufferChannel ;
+
+/** StateManagement for a number of long values, a common need
+ * (might as well store ints as long on disk
+ * for small numbers of integers)
+ */
+public class StateMgrData extends StateMgrBase {
+ private final long[] data ;
+
+ public StateMgrData(BufferChannel storage, long... initialData) {
+ super(storage, numBytes(initialData)) ;
+ data = copy(initialData) ;
+ super.init() ;
+ }
+
+ @Override
+ protected void init() { throw new TransactionException("Don't call init()") ; }
+
+ private static long[] copy(long[] data) { return Arrays.copyOf(data, data.length) ; }
+
+ private static int numBytes(long[] data) { return data.length * Long.BYTES ; }
+
+ // Protected - leave whether to expose these operations as "public"
+ // to the subclass. A subclass may choose instead to make these
+ // to more meaningful names, or ensure that daat consistentecny rules are applied.
+
+ protected long[] getData() {
+ return copy(data) ;
+ }
+
+ protected void setData(long... newData) {
+ if ( newData.length != data.length )
+ throw new IllegalArgumentException() ;
+ System.arraycopy(newData, 0, data, 0, data.length);
+ }
+
+ protected long get(int i) {
+ return data[i] ;
+ }
+
+ protected void set(int i, long v) {
+ data[i] = v ;
+ super.setDirtyFlag() ;
+ }
+
+ @Override
+ protected ByteBuffer serialize(ByteBuffer bytes) {
+ for ( int i = 0 ; i < data.length ; i++ )
+ bytes.putLong(data[i]) ;
+ return bytes ;
+ }
+
+ @Override
+ protected void deserialize(ByteBuffer bytes) {
+ for ( int i = 0 ; i < data.length ; i++ )
+ data[i] = bytes.getLong() ;
+ }
+
+ @Override
+ protected void writeStateEvent() {}
+
+ @Override
+ protected void readStateEvent() {}
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrDataIdx.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrDataIdx.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrDataIdx.java
new file mode 100644
index 0000000..57120c2
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/StateMgrDataIdx.java
@@ -0,0 +1,45 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import org.seaborne.dboe.base.file.BufferChannel ;
+
+/** StateManagement for a number of long values,
+ * state accessed by index.
+ */
+public final class StateMgrDataIdx extends StateMgrData {
+ public StateMgrDataIdx(BufferChannel storage, long... initialData) {
+ super(storage, initialData) ;
+ }
+
+ // Expose these operations as public
+
+ @Override
+ public long[] getData() { return super.getData() ; }
+
+ @Override
+ public void setData(long... newData) { super.setData(newData); }
+
+ @Override
+ public long get(int i) { return super.get(i) ; }
+
+ @Override
+ public void set(int i, long v) { super.set(i, v); }
+}
+
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTrans.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTrans.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTrans.java
new file mode 100644
index 0000000..c33a956
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTrans.java
@@ -0,0 +1,53 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import java.nio.ByteBuffer ;
+
+/** Single component aspect of a transaction */
+final
+class SysTrans {
+ private final TransactionalComponent elt ;
+ private final Transaction transaction ;
+ private final TxnId txnId ;
+
+ public SysTrans(TransactionalComponent elt, Transaction transaction, TxnId txnId) {
+ this.elt = elt ;
+ this.transaction = transaction ;
+ this.txnId = txnId ;
+ }
+
+ public void begin() { }
+ public boolean promote() { return elt.promote(transaction) ; }
+
+ public ByteBuffer commitPrepare() { return elt.commitPrepare(transaction) ; }
+
+ public void commit() { elt.commit(transaction); }
+
+ public void commitEnd() { elt.commitEnd(transaction); }
+
+ public void abort() { elt.abort(transaction); }
+
+ public void complete() { elt.complete(transaction); }
+
+ public Transaction getTransaction() { return transaction ; }
+ public TxnId getTxnId() { return txnId ; }
+ public TransactionalComponent getComponent() { return elt ; }
+ public ComponentId getComponentId() { return elt.getComponentId() ; }
+}
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTransState.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTransState.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTransState.java
new file mode 100644
index 0000000..975e5ee
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/SysTransState.java
@@ -0,0 +1,36 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+/** State for one detached components. */
+public class SysTransState {
+ private final TransactionalComponent elt ;
+ private final Transaction transaction ;
+ private final Object state ;
+
+ public SysTransState(TransactionalComponent elt, Transaction transaction, Object state) {
+ this.elt = elt ;
+ this.transaction = transaction ;
+ this.state = state ;
+ }
+
+ public TransactionalComponent getComponent() { return elt ; }
+ public Transaction getTransaction() { return transaction ; }
+ public Object getState() { return state ; }
+}
http://git-wip-us.apache.org/repos/asf/jena/blob/1dabea3a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/Transaction.java
----------------------------------------------------------------------
diff --git a/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/Transaction.java b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/Transaction.java
new file mode 100644
index 0000000..3feccdd
--- /dev/null
+++ b/jena-db/jena-dboe-transaction/src/main/java/org/seaborne/dboe/transaction/txn/Transaction.java
@@ -0,0 +1,295 @@
+/*
+ * 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.seaborne.dboe.transaction.txn;
+
+import static org.seaborne.dboe.transaction.txn.TxnState.ABORTED;
+import static org.seaborne.dboe.transaction.txn.TxnState.ACTIVE;
+import static org.seaborne.dboe.transaction.txn.TxnState.COMMIT;
+import static org.seaborne.dboe.transaction.txn.TxnState.COMMITTED;
+import static org.seaborne.dboe.transaction.txn.TxnState.DETACHED;
+import static org.seaborne.dboe.transaction.txn.TxnState.END_ABORTED;
+import static org.seaborne.dboe.transaction.txn.TxnState.END_COMMITTED;
+import static org.seaborne.dboe.transaction.txn.TxnState.INACTIVE;
+import static org.seaborne.dboe.transaction.txn.TxnState.PREPARE;
+
+import java.util.List ;
+import java.util.Objects ;
+import java.util.concurrent.atomic.AtomicReference ;
+
+import org.apache.jena.query.ReadWrite ;
+
+/**
+ * A transaction as the composition of actions on components.
+ * Works in conjunction with the TransactionCoordinator
+ * to provide the transaction lifecycle.
+ * @see TransactionCoordinator
+ * @see TransactionalComponent
+ */
+final
+public class Transaction implements TransactionInfo {
+ // Using an AtomicReference<TxnState> requires that
+ // TransactionalComponentLifecycle.internalComplete
+ // frees the thread local for the threadTxn, otherwise memory
+ // usage grows. If a plain member variable is used slow growth
+ // is still seen.
+ // Nulling txnMgr and clearing components stops that slow growth.
+
+ private TransactionCoordinator txnMgr ;
+ private final TxnId txnId ;
+ private final List<SysTrans> components ;
+
+ // Using an AtomicReference makes this observable from the outside.
+ // It also allow for multithreaded transactions (later).
+ private final AtomicReference<TxnState> state = new AtomicReference<>() ;
+ //private TxnState state ;
+ private long dataVersion ;
+ private ReadWrite mode ;
+
+ public Transaction(TransactionCoordinator txnMgr, TxnId txnId, ReadWrite readWrite, long dataVersion, List<SysTrans> components) {
+ Objects.requireNonNull(txnMgr) ;
+ Objects.requireNonNull(txnId) ;
+ Objects.requireNonNull(readWrite) ;
+ Objects.requireNonNull(components) ;
+ this.txnMgr = txnMgr ;
+ this.txnId = txnId ;
+ this.mode = readWrite ;
+ this.dataVersion = dataVersion ;
+ this.components = components ;
+ setState(INACTIVE) ;
+ }
+
+ /*package*/ void resetDataVersion(long dataVersion) {
+ this.dataVersion = dataVersion;
+ }
+
+ /*package*/ void setState(TxnState newState) {
+ state.set(newState) ;
+ }
+
+ @Override
+ public TxnState getState() {
+ return state.get() ;
+ }
+
+ /**
+ * Each transaction is allocated a serialization point by the transaction
+ * coordinator. Normally, this is related to this number and it increases
+ * over time as the data changes. Two readers can have the same
+ * serialization point - they are working with the same view of the data.
+ */
+ @Override
+ public long getDataVersion() {
+ return dataVersion ;
+ }
+
+ public void begin() {
+ checkState(INACTIVE);
+ components.forEach((c) -> c.begin()) ;
+ setState(ACTIVE) ;
+ }
+
+ public boolean promote() {
+ checkState(ACTIVE);
+ boolean b = txnMgr.promoteTxn(this) ;
+ if ( !b )
+ return false ;
+ mode = ReadWrite.WRITE;
+ return true ;
+ }
+
+ /*package*/ void promoteComponents() {
+ // Call back from the Transaction coordinator during promote.
+ components.forEach((c) -> {
+ if ( ! c.promote() )
+ throw new TransactionException("Failed to promote") ;
+ }) ;
+ mode = ReadWrite.WRITE ;
+ }
+
+ public void notifyUpdate() {
+ checkState(ACTIVE) ;
+ if ( mode == ReadWrite.READ ) {
+ System.err.println("notifyUpdate - promote needed") ;
+ promote() ;
+ mode = ReadWrite.WRITE ;
+ }
+ }
+
+ public void prepare() {
+ checkState(ACTIVE) ;
+ if ( mode == ReadWrite.WRITE )
+ txnMgr.executePrepare(this) ;
+ setState(PREPARE);
+ }
+
+ public void commit() {
+ // Split into READ and WRITE forms.
+ TxnState s = getState();
+ if ( s == ACTIVE )
+ // Auto exec prepare().
+ prepare() ;
+ checkState(PREPARE) ;
+ setState(COMMIT) ;
+ switch(mode) {
+ case WRITE:
+ txnMgr.executeCommit(this,
+ ()->{components.forEach((c) -> c.commit()) ; } ,
+ ()->{components.forEach((c) -> c.commitEnd()) ; } ) ;
+ break ;
+ case READ:
+ // Different lifecycle?
+ txnMgr.executeCommit(this,
+ ()->{components.forEach((c) -> c.commit()) ; } ,
+ ()->{components.forEach((c) -> c.commitEnd()) ; } ) ;
+ break ;
+ }
+ setState(COMMITTED) ;
+ }
+
+ public void abort() {
+ abort$();
+ endInternal() ;
+ }
+
+ private void abort$() {
+ // Split into READ and WRITE forms.
+ checkState(ACTIVE, ABORTED) ;
+ // Includes notification start/finish
+ txnMgr.executeAbort(this, ()-> { components.forEach((c) -> c.abort()) ; }) ;
+ setState(ABORTED) ;
+ }
+
+ public void end() {
+ txnMgr.notifyEndStart(this) ;
+ if ( isWriteTxn() && getState() == ACTIVE ) {
+ throw new TransactionException("Write transaction with no commit or abort") ;
+ //Log.warn(this, "Write transaction with no commit() or abort() before end()");
+ // Just the abort process.
+ //abort$() ;
+ }
+ endInternal() ;
+ txnMgr.notifyEndFinish(this) ;
+ txnMgr = null ;
+ //components.clear() ;
+ }
+
+ private void endInternal() {
+ if ( hasFinalised() )
+ return ;
+ // Called once, at the first abort/commit/end.
+ txnMgr.notifyCompleteStart(this);
+ components.forEach((c) -> c.complete()) ;
+ txnMgr.completed(this) ;
+ if ( getState() == COMMITTED )
+ setState(END_COMMITTED);
+ else
+ setState(END_ABORTED);
+ txnMgr.notifyCompleteFinish(this);
+ }
+
+ /*package*/ List<SysTrans> getComponents() {
+ return components ;
+ }
+
+ /*package*/ void detach() {
+ checkState(ACTIVE,PREPARE) ;
+ setState(DETACHED) ;
+ }
+
+ /*package*/ void attach() {
+ checkState(DETACHED) ;
+ setState(ACTIVE) ;
+ }
+
+ public void requireWriteTxn() {
+ checkState(ACTIVE) ;
+ if ( mode != ReadWrite.WRITE )
+ throw new TransactionException("Not a write transaction") ;
+ }
+
+ @Override
+ public boolean hasStarted() {
+ TxnState x = getState() ;
+ return x == INACTIVE ;
+ }
+
+ @Override
+ public boolean hasFinished() {
+ TxnState x = getState() ;
+ return x == COMMITTED || x == ABORTED || x == END_COMMITTED || x == END_ABORTED ;
+ }
+
+ @Override
+ public boolean hasFinalised() {
+ TxnState x = getState() ;
+ return x == END_COMMITTED || x == END_ABORTED ;
+ }
+
+ @Override
+ public TxnId getTxnId() { return txnId ; }
+
+ @Override
+ public ReadWrite getMode() { return mode ; }
+
+ /** Is this a READ transaction?
+ * Convenience operation equivalent to {@code (getMode() == READ)}
+ */
+ @Override
+ public boolean isReadTxn() { return mode == ReadWrite.READ ; }
+
+ /** Is this a WRITE transaction?
+ * Convenience operation equivalent to {@code (getMode() == WRITE)}
+ */
+ @Override
+ public boolean isWriteTxn() { return mode == ReadWrite.WRITE ; }
+
+ // hashCode/equality
+ // These must be object equality. No two transactions objects are .equals unless they are ==
+
+ private void checkWriteTxn() {
+ if ( ! isActiveTxn() || ! isWriteTxn() )
+ throw new TransactionException("Not in a write transaction") ;
+ }
+
+ // XXX Duplicate -- TransactionalComponentLifecycle
+ private void checkState(TxnState expected) {
+ TxnState s = getState();
+ if ( s != expected )
+ throw new TransactionException("Transaction is in state "+s+": expected state "+expected) ;
+ }
+
+ private void checkState(TxnState expected1, TxnState expected2) {
+ TxnState s = getState();
+ if ( s != expected1 && s != expected2 )
+ throw new TransactionException("Transaction is in state "+s+": expected state "+expected1+" or "+expected2) ;
+ }
+
+ // Avoid varargs ... undue worry?
+ private void checkState(TxnState expected1, TxnState expected2, TxnState expected3) {
+ TxnState s = getState();
+ if ( s != expected1 && s != expected2 && s != expected3 )
+ throw new TransactionException("Transaction is in state "+s+": expected state "+expected1+", "+expected2+" or "+expected3) ;
+ }
+
+ @Override
+ public boolean isActiveTxn() {
+ return getState() != INACTIVE ;
+ }
+}
+