--- /dev/null
+/*******************************************************************************\r
+ * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
+ * in Industry THTH ry.\r
+ * All rights reserved. This program and the accompanying materials\r
+ * are made available under the terms of the Eclipse Public License v1.0\r
+ * which accompanies this distribution, and is available at\r
+ * http://www.eclipse.org/legal/epl-v10.html\r
+ *\r
+ * Contributors:\r
+ * VTT Technical Research Centre of Finland - initial API and implementation\r
+ *******************************************************************************/\r
+package fi.vtt.simantics.procore.internal;\r
+\r
+import java.io.IOException;\r
+import java.util.List;\r
+import java.util.TreeMap;\r
+import java.util.concurrent.atomic.AtomicLong;\r
+\r
+import org.simantics.databoard.Bindings;\r
+import org.simantics.databoard.binding.impl.TreeMapBinding;\r
+import org.simantics.databoard.serialization.SerializationException;\r
+import org.simantics.databoard.serialization.Serializer;\r
+import org.simantics.db.Operation;\r
+import org.simantics.db.Session;\r
+import org.simantics.db.common.CommentMetadata;\r
+import org.simantics.db.common.CommitMetadata;\r
+import org.simantics.db.common.MetadataUtils;\r
+import org.simantics.db.common.utils.Logger;\r
+import org.simantics.db.exception.DatabaseException;\r
+import org.simantics.db.exception.ValidationException;\r
+import org.simantics.db.impl.graph.WriteSupport;\r
+import org.simantics.db.procore.protocol.Constants;\r
+import org.simantics.db.request.WriteTraits;\r
+import org.simantics.db.service.ClusterSetsSupport;\r
+import org.simantics.db.service.ClusterUID;\r
+import org.simantics.db.service.ExternalOperation;\r
+import org.simantics.db.service.TransactionPolicySupport;\r
+\r
+public class TransactionToken implements GraphSession.Listener {\r
+ private boolean DEBUG = false;\r
+ private Session session;\r
+ private GraphSession graphSession;\r
+ private AtomicLong id;\r
+ // This is the last change set that we have send commit for.\r
+ // This does not mean that those commits have been accepted in server.\r
+ private long lastChangeSetId;\r
+ private OperationImpl lastOperation = null;\r
+ enum Type { Null, Read, Write }\r
+ private Type type;\r
+ // Count of pending transactions.\r
+ private int pending = 0;\r
+ TransactionToken(TransactionPolicySupport tps, Session session, GraphSession graphSession, TransactionToken old) {\r
+ this.session = session;\r
+ this.graphSession = graphSession;\r
+ this.type = Type.Null;\r
+ this.id = new AtomicLong();\r
+ this.id.set(Constants.NullTransactionId);\r
+ long firstChangeSetId = graphSession.getFirstChangeSetId();\r
+ if (null != old && old.lastChangeSetId > firstChangeSetId)\r
+ this.lastChangeSetId = old.lastChangeSetId;\r
+ else\r
+ this.lastChangeSetId = firstChangeSetId;\r
+ graphSession.setListener(this);\r
+ }\r
+ void startReadTransaction(int thread)\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Start read transaction " + graphSession);\r
+ if (Type.Null == type) {\r
+ id.set(graphSession.askReadTransaction(thread));\r
+ if (DEBUG)\r
+ if (++pending != 1)\r
+ System.out.println("kraa: read beg pending=" + pending);\r
+ type = Type.Read;\r
+ }\r
+ }\r
+ void stopReadTransaction()\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Stop read transaction " + graphSession);\r
+ if (Type.Null == type)\r
+ return;\r
+ endTransaction(false);\r
+ if (DEBUG)\r
+ if (--pending != 0)\r
+ System.out.println("kraa: read end pending=" + pending);\r
+ }\r
+ void startWriteTransaction(int thread)\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Start write transaction " + graphSession);\r
+ if (Type.Null == type) {\r
+ id.set(graphSession.askWriteTransaction(thread, Constants.NullTransactionId));\r
+ if (DEBUG)\r
+ if (++pending != 1)\r
+ System.out.println("kraa: write beg pending=" + pending);\r
+ } else if (Type.Read == type) {\r
+ id.set(graphSession.askWriteTransaction(thread, id.get()));\r
+ if (DEBUG)\r
+ if (++pending != 1)\r
+ System.out.println("kraa: write beg pending=" + pending);\r
+ }\r
+ type = Type.Write;\r
+ }\r
+ void cancelBegin(WriteSupport writeSupport, SynchronizeContextI context, ClusterStream clusterStream)\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: CancelBegin " + graphSession);\r
+ if (Type.Null == type)\r
+ return;\r
+ else if (Type.Read == type)\r
+ throw new ValidationException("Illegal token type.");\r
+ TreeMap<String, byte[]> metadata = writeSupport.getMetadata();\r
+ boolean empty = clusterStream.reallyFlush();\r
+ if (empty) {\r
+ boolean noMetadata = !hasMetadata(metadata);\r
+ if (noMetadata)\r
+ return;\r
+ }\r
+ byte[] data = makeContext(null, metadata);\r
+ graphSession.cancelCommit(id.get(), lastChangeSetId, data, context);\r
+ ++lastChangeSetId;\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: CancelBegin cancelled cs=" + lastChangeSetId);\r
+ clusterStream.accept();\r
+ writeSupport.commitDone(null, lastChangeSetId);\r
+ }\r
+ void cancelEnd(WriteSupport writeSupport, WriteTraits writeTraits, ClusterStream clusterStream)\r
+ throws DatabaseException {\r
+ if (clusterStream.reallyFlush())\r
+ return;\r
+ TreeMap<String, byte[]> metadata = writeSupport.getMetadata();\r
+ byte[] data = makeContext(null, metadata);\r
+ graphSession.acceptCommit(id.get(), lastChangeSetId, data);\r
+ ++lastChangeSetId;\r
+ if (DEBUG)\r
+ System.out.println("DEBUG CancelEnd accepted commit cs=" + lastChangeSetId);\r
+ clusterStream.accept();\r
+ writeSupport.commitDone(writeTraits, lastChangeSetId);\r
+ }\r
+ void setCombine(boolean a) {\r
+ combine = a;\r
+ }\r
+ private boolean hasMetadata(TreeMap<String, byte[]> metadata) {\r
+ \r
+ if(metadata == null)\r
+ return false;\r
+ if(metadata.size() == 0)\r
+ return false;\r
+ if(metadata.size() == 1 && MetadataUtils.hasMetadata(metadata, CommentMetadata.class))\r
+ return false;\r
+ \r
+ return true;\r
+ \r
+ }\r
+ private boolean combine = false;\r
+ void commitWriteTransaction(WriteSupport writeSupport, WriteTraits writeTraits, ClusterStream clusterStream, Operation dummy)\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Commit write transaction " + graphSession);\r
+ if (Type.Null == type)\r
+ return;\r
+ else if (Type.Read == type)\r
+ throw new ValidationException("Illegal transaction type.");\r
+ else if (id.get() == Constants.NullTransactionId)\r
+ throw new ValidationException("Illegal transaction id.");\r
+ TreeMap<String, byte[]> metadata = writeSupport.getMetadata();\r
+ boolean empty = clusterStream.reallyFlush();\r
+ if (empty) {\r
+ boolean noMetadata = !hasMetadata(metadata);\r
+ if (noMetadata) {\r
+ List<ExternalOperation> ext = graphSession.undoContext.getPendingExternals();\r
+ if(!ext.isEmpty()) {\r
+ CommentMetadata cm = writeSupport.getMetadata(CommentMetadata.class);\r
+ writeSupport.addMetadata(cm.add("Automatically generated comment for empty commit with external undo operation(s)."));\r
+ metadata = writeSupport.getMetadata();\r
+ Logger.defaultLogError("A generated comment was added into a commit with only some external undo operation(s).");\r
+ } else {\r
+ graphSession.undoContext.cancelCommit();\r
+ return;\r
+ }\r
+ }\r
+ }\r
+\r
+ CommentMetadata cm = writeSupport.getMetadata(CommentMetadata.class);\r
+ if(cm.size() == 0)\r
+ writeSupport.addMetadata(cm.add(writeTraits.toString()));\r
+\r
+ metadata = writeSupport.getMetadata();\r
+\r
+ Operation op = combine ? lastOperation : null;\r
+ combine = true;\r
+ byte[] data = makeContext(op, metadata);\r
+// new Exception("committing cs + " + lastChangeSetId + " with metadata " + data).printStackTrace();\r
+ graphSession.acceptCommit(id.get(), lastChangeSetId, data);\r
+ ++lastChangeSetId;\r
+ if (DEBUG)\r
+ System.out.println("DEBUG Accpeted commit cs=" + lastChangeSetId);\r
+ clusterStream.accept();\r
+ writeSupport.commitDone(writeTraits, lastChangeSetId);\r
+ long opid = (null == op) ? lastChangeSetId : op.getId();\r
+ lastOperation = new OperationImpl(opid, lastChangeSetId, graphSession.undoContext.getPendingExternals());\r
+ graphSession.undoContext.commitOk(lastOperation);\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Accepted operation id=" + lastOperation.getId() + " cs=" + lastOperation.getCSId() + ".");\r
+ }\r
+ void stopWriteTransaction()\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Stop write transaction begin " + graphSession);\r
+ if (Type.Null == type)\r
+ return;\r
+ else if (Type.Read == type)\r
+ throw new ValidationException("Illegal transaction type.");\r
+ // Note that even if we do cancel the cluster sets are not restored.\r
+ try {\r
+ session.getService(ClusterSetsSupport.class).save();\r
+ } catch (Exception e) {\r
+ e.printStackTrace();\r
+ Logger.defaultLogError("Failed to save cluster sets.", e);\r
+ }\r
+ graphSession.undoContext.cancelCommit();\r
+ endTransaction(true);\r
+ if (DEBUG)\r
+ if (--pending != 0)\r
+ System.out.println("kraa: write end pending=" + pending);\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Stop write transaction end " + graphSession);\r
+ }\r
+ void closing()\r
+ throws DatabaseException {\r
+ if (Type.Null == type || id.get() == Constants.NullTransactionId)\r
+ return;\r
+ endTransaction(true);\r
+ if (DEBUG)\r
+ if (--pending != 0)\r
+ System.out.println("kraa: closing pending=" + pending);\r
+ }\r
+ void close() {\r
+ if (id.get() != Constants.NullTransactionId)\r
+ try {\r
+ endTransaction(true);\r
+ if (DEBUG)\r
+ if (--pending != 0)\r
+ System.out.println("kraa: closing pending=" + pending);\r
+ } catch (DatabaseException e) {\r
+ if (DEBUG) {\r
+ System.out.println("Close did not finish cleanly.");\r
+ e.printStackTrace();\r
+ }\r
+ }\r
+ }\r
+ public Operation getLastOperation() {\r
+ return lastOperation;\r
+ }\r
+ public long getHeadRevisionId() {\r
+ return lastChangeSetId;\r
+ }\r
+\r
+ static Serializer METADATA_SERIALIZER =\r
+ Bindings.getSerializerUnchecked( new TreeMapBinding(Bindings.STRING,\r
+ Bindings.BYTE_ARRAY) );\r
+\r
+ private byte[] makeContext(Operation op, TreeMap<String, byte[]> properties) {\r
+ if (properties == null)\r
+ properties = new TreeMap<String, byte[]>();\r
+ long csid = (null != op) ? op.getId() : 0;\r
+ properties.put(CommitMetadata.class.getName(), new CommitMetadata(csid).serialise(session));\r
+ properties.put("opid", Long.toString(csid).getBytes());\r
+\r
+ try {\r
+ return METADATA_SERIALIZER.serialize(properties);\r
+ } catch (SerializationException e) {\r
+ e.printStackTrace();\r
+ return new byte[0];\r
+ } catch (IOException e) {\r
+ e.printStackTrace();\r
+ return new byte[0];\r
+ }\r
+ }\r
+ private void endTransaction(boolean write)\r
+ throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: TransactionToken.end begin " + graphSession);\r
+ try {\r
+ graphSession.endTransaction(id.get(), write);\r
+ } finally {\r
+ type = Type.Null;\r
+ id.set(Constants.NullTransactionId);\r
+ synchronized (this) {\r
+ this.notify();\r
+ }\r
+ }\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: TransactionToken.end end " + graphSession);\r
+ }\r
+ private void updateLastChangeSetId(long csid) {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: TransactionToken: Update last cs=" + lastChangeSetId + " new=" + csid + " "\r
+ + graphSession + " " + this);\r
+ if (lastChangeSetId > csid)\r
+ return; // throw new\r
+// InternalError("Change set id has been corrupted. Last cs=" + lastChangeSetId\r
+// + " new=" + csid);\r
+ lastChangeSetId = csid;\r
+ }\r
+ public static void testEmpty() {\r
+ final int REPEAT_COUNT = 20;\r
+ final int REQUEST_COUNT = 500; // 1000;\r
+ double totalTime = 0;\r
+ for (int j = 0; j < REPEAT_COUNT; ++j) {\r
+ long start = System.nanoTime();\r
+ for (int i = 0; i < REQUEST_COUNT; ++i) {\r
+ }\r
+ long end = System.nanoTime();\r
+ double time = (end - start) * (1e-9);\r
+ totalTime += time;\r
+ }\r
+ double speed = REPEAT_COUNT * REQUEST_COUNT / totalTime;\r
+ String t = "Speed was " + speed + " ops per second.";\r
+ System.out.println(t);\r
+ }\r
+ @Override\r
+ public void onChangeSetId(int thread, long csid, boolean refresh) {\r
+ long ocsid = lastChangeSetId;\r
+ updateLastChangeSetId(csid);\r
+ if (refresh && csid > ocsid && id.get() == Constants.NullTransactionId)\r
+ refresh(thread, ocsid);\r
+ }\r
+ private void refresh(int thread, long csid) {\r
+ try {\r
+ if (session instanceof SessionImplSocket) {\r
+ ClusterUID[] clusterUID = graphSession.getRefresh2(csid);\r
+ ((SessionImplSocket)session).refresh(thread, clusterUID, csid);\r
+ }\r
+ } catch (DatabaseException e) {\r
+ Logger.defaultLogError(e);\r
+ }\r
+ }\r
+}
\ No newline at end of file
Code Review / simantics / platform.git / blobdiff